From 97a0c90c23d709404a6dbc90adedfadce85a46a6 Mon Sep 17 00:00:00 2001 From: sdairs Date: Fri, 7 Aug 2026 23:15:46 +0100 Subject: [PATCH] Extract ClickPipes runtime and finalize cloud CLI split --- AGENTS.md | 24 +- crates/clickhousectl/src/cloud/cli.rs | 16 +- crates/clickhousectl/src/cloud/clickpipes.rs | 1694 ++++++++++++++++++ crates/clickhousectl/src/cloud/client.rs | 139 +- crates/clickhousectl/src/cloud/commands.rs | 1235 ------------- crates/clickhousectl/src/cloud/mod.rs | 183 +- 6 files changed, 1718 insertions(+), 1573 deletions(-) delete mode 100644 crates/clickhousectl/src/cloud/commands.rs diff --git a/AGENTS.md b/AGENTS.md index 16d4872a..e86fc55d 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -10,7 +10,7 @@ This is a Cargo workspace with two crates: The user-facing CLI surface. Contains all logic for local commands, wraps `clickhouse-cloud-api` for cloud. -- Cloud handlers go through the `CloudClient` wrapper (`src/cloud/client.rs`), not `clickhouse_cloud_api::Client` directly. The wrapper handles credential precedence, error conversion, and response unwrapping. +- Cloud handlers go through `CloudClient` wrapper methods co-located in each domain module, not `clickhouse_cloud_api::Client` directly. `src/cloud/client.rs` owns the core client, credential precedence, error conversion, and response unwrapping. - Cloud handlers always support `--json` output unless there is good reason not to. JSON is emitted automatically when `--json` is passed or a coding agent is detected (`is_ai_agent::detect()` via the `json_output()` helper in `main.rs`). - `CloudError` carries a `kind: CloudErrorKind` (`Auth` for 401/403 and missing credentials, else `Generic`). It maps to `Error::AuthRequired` / `Error::Cloud` in `cloud::run`. Dispatched commands exit with `0` on success or use `Error::exit_code()` for failures: `1` error, `3` cancelled, `4` auth required. Clap uses `2` for usage errors. @@ -22,7 +22,7 @@ The CLI does not need to have 100% coverage of endpoints exposed by the API libr #### Adding a command -For both local and cloud commands, define the clap variant in the appropriate `cli.rs`, then wire dispatch in the owning runtime module. +Local clap definitions live in `src/local/cli.rs`. Cloud clap definitions, handlers, builders, wrapper methods, dispatch, and tests are co-located in the owning domain module under `src/cloud/`; `src/cloud/cli.rs` contains only the top-level cloud arguments and command enum. **Local subcommand:** @@ -32,13 +32,13 @@ For both local and cloud commands, define the clap variant in the appropriate `c **Cloud subcommand:** -1. Make sure `clickhouse-cloud-api` has already been updated to support necessary endpoints & models. -2. Add the variant to the relevant sub-enum in `src/cloud/cli.rs` (or `src/cloud/postgres.rs` for Postgres). Create a new sub-enum if the surface warrants its own grouping. -3. Classify the new variant in `CloudCommands::is_write_command()` in `src/cloud/cli.rs` (Postgres variants go in the equivalent `is_write()` on the Postgres enum). OAuth (Bearer) auth is read-only; write commands require API key auth and we fail fast on OAuth + write. The match has no wildcards, so the compiler will reject a missing arm — but you still need to make the read/write call deliberately, and add a case to both the `is_write_command_read_only_commands` and `is_write_command_destructive_commands` tests. -4. Add non-Postgres match arms to `cloud::dispatch()` in `src/cloud/mod.rs`. Postgres dispatch belongs in `src/cloud/postgres.rs::run`; if another domain later gains its own runtime dispatcher, have `cloud::dispatch()` delegate to it rather than keeping that domain's match arms centrally. -5. Add a thin wrapper method on `CloudClient` in `src/cloud/client.rs`. It should delegate to `self.api().()`, map errors via `self.convert_error(e)`, and unwrap with `Self::unwrap_response`. Use the library's request/response types here. -6. If the command sends a request body, extract a `build__request(...)` helper in `src/cloud/commands.rs` that returns the library's request struct. Cover the helper with minimal + maximal unit tests in the `mod tests` block at the bottom of `commands.rs`, asserting directly on library struct fields. -7. Implement the handler in `src/cloud/commands.rs`. For body-sending commands the handler calls the build helper, passes the result through the `CloudClient` wrapper, and prints with the `--json` output pattern. For detail/get views (rendering a single resource), drive human output through `print_human` so it shares serde's behaviour — including deprecated-field hiding — instead of hand-writing `println!` lines: +1. Make sure `clickhouse-cloud-api` has already been updated to support necessary endpoints and models. +2. Add the clap variant and argument structs to the owning `src/cloud/.rs` module. Create a new domain module and privately re-export its command enum from `src/cloud/cli.rs` if the surface warrants its own grouping. +3. Classify the variant in the domain command enum's exhaustive `is_write()` match. OAuth (Bearer) auth is read-only; write commands require API key auth and fail fast on OAuth + write. `CloudCommands::is_write_command()` in `src/cloud/cli.rs` exhaustively delegates to each domain. Add read/write classification tests next to the domain clap definitions. +4. Add the exhaustive command match to the domain's `run()` dispatcher. `cloud::dispatch()` in `src/cloud/mod.rs` delegates only at the top-level `CloudCommands` boundary; add one delegation arm there only when introducing a new domain. +5. Add a thin wrapper method in the domain module's `impl CloudClient` block. It should delegate to `self.api().()`, map errors via `self.convert_error(e)` or `self.convert_error_for_organization(e, org_id)`, and unwrap with `Self::unwrap_response`. Use the library's request/response types here. +6. If the command sends a request body, extract a `build__request(...)` helper in the same domain module that returns the library's request struct. Cover the helper with minimal + maximal unit tests in that module's `mod tests`, asserting directly on library struct fields. +7. Implement the handler in the same domain module. For body-sending commands the handler calls the build helper, passes the result through the `CloudClient` wrapper, and prints with the `--json` output pattern. For detail/get views (rendering a single resource), drive human output through `print_human` so it shares serde's behaviour — including deprecated-field hiding — instead of hand-writing `println!` lines: ```rust if json { println!("{}", serde_json::to_string_pretty(&data)?); @@ -49,7 +49,7 @@ For both local and cloud commands, define the clap variant in the appropriate `c List views stay as `tabled` tables, and short action confirmations (e.g. "Service X starting") stay as plain `println!`. Every field of a library response type is `Option` (see Request and response models), so never `unwrap()`/`expect()` one. Render absence with `crate::cloud::output::or_absent` (`-`) or `ABSENT` in `tabled` cells and plain output, and have `--filter`-style predicates treat an absent field as non-matching. `print_human` and `--json` serialize the model and need no per-field work. -8. Add `Cli::try_parse_from` coverage in `src/cloud/cli.rs` for the new command's body-related flags, asserting parsed values. +8. Add `Cli::try_parse_from` coverage next to the domain command definition for the new command's body-related flags, asserting parsed values. ### API library (`crates/clickhouse-cloud-api/`) @@ -186,8 +186,8 @@ Real cloud integration tests, 100% OpenAPI spec coverage. Cost is not a reason t ### clickhousectl CLI -- **Clap parsing** — `Cli::try_parse_from` tests next to each command definition (`src/cli.rs`, `src/cloud/cli.rs`, `src/cloud/postgres.rs`, `src/local/cli.rs`). Assert flag names, types, defaults, and repeatability. -- **Request builders** — unit tests for `build_*_request` helpers in `src/cloud/commands.rs`, asserting on library request-struct fields with minimal + maximal inputs. +- **Clap parsing** — `Cli::try_parse_from` tests next to each command definition (`src/cli.rs`, the owning `src/cloud/.rs`, and `src/local/cli.rs`). Assert flag names, types, defaults, and repeatability. +- **Request builders** — unit tests for `build_*_request` helpers next to the owning cloud domain code, asserting on library request-struct fields with minimal + maximal inputs. - **Subprocess + wiremock** — `tests/cli_request_shape_test.rs`. Spawn the real binary against a local mock server and assert on the recorded request JSON. Used when the handler has runtime behavior beyond struct construction (file reads, base64 encoding, etc.) — currently ClickPipes. - **Pure logic** — inline `mod tests` blocks across `src/` for version resolution, auth precedence, output formatting, platform detection, and other module-local helpers. diff --git a/crates/clickhousectl/src/cloud/cli.rs b/crates/clickhousectl/src/cloud/cli.rs index 628dd950..9ec31a79 100644 --- a/crates/clickhousectl/src/cloud/cli.rs +++ b/crates/clickhousectl/src/cloud/cli.rs @@ -1,18 +1,20 @@ -pub use crate::cloud::activity::ActivityCommands; -pub use crate::cloud::api_keys::KeyCommands; -pub use crate::cloud::auth::AuthCommands; +pub(crate) use crate::cloud::activity::ActivityCommands; +pub(crate) use crate::cloud::api_keys::KeyCommands; +pub(crate) use crate::cloud::auth::AuthCommands; #[allow(unused_imports)] -pub use crate::cloud::backups::{BackupCommands, BackupConfigCommands}; +pub(crate) use crate::cloud::backups::{BackupCommands, BackupConfigCommands}; #[allow(unused_imports)] -pub use crate::cloud::clickpipes::{ +pub(crate) use crate::cloud::clickpipes::{ BigQueryCreateArgs, ClickPipeCommands, ClickPipeCreateCommands, ClickPipeSchemaDiscoverCommands, ClickPipeSettingsCommands, KafkaCreateArgs, KafkaSourceFields, KinesisCreateArgs, KinesisSourceFields, MongoDbCreateArgs, MySqlCreateArgs, ObjectStorageCreateArgs, PostgresCreateArgs, }; -pub use crate::cloud::organizations::{InvitationCommands, MemberCommands, OrgCommands}; +pub(crate) use crate::cloud::organizations::{InvitationCommands, MemberCommands, OrgCommands}; #[allow(unused_imports)] -pub use crate::cloud::services::{PrivateEndpointCommands, QueryEndpointCommands, ServiceCommands}; +pub(crate) use crate::cloud::services::{ + PrivateEndpointCommands, QueryEndpointCommands, ServiceCommands, +}; use clap::{Args, Subcommand}; #[derive(Args)] diff --git a/crates/clickhousectl/src/cloud/clickpipes.rs b/crates/clickhousectl/src/cloud/clickpipes.rs index 07915fb0..ab56ad9e 100644 --- a/crates/clickhousectl/src/cloud/clickpipes.rs +++ b/crates/clickhousectl/src/cloud/clickpipes.rs @@ -1,5 +1,9 @@ +use crate::cloud::client::CloudClient; +use crate::cloud::output::{or_absent, print_human}; +use crate::cloud::shared::resolve_org_id; use clap::builder::PossibleValuesParser; use clap::{Args, Subcommand}; +use tabled::{Table, Tabled, settings::Style}; // Valid wire values for each ClickPipe enum the CLI accepts as a string argument. // Kept in sync with the clickhouse-cloud-api library enums; extra variants are @@ -907,6 +911,1292 @@ pub struct BigQueryCreateArgs { pub org_id: Option, } +pub async fn run( + client: &CloudClient, + command: ClickPipeCommands, + json: bool, +) -> Result<(), Box> { + match command { + ClickPipeCommands::List { service_id, org_id } => { + clickpipe_list(client, &service_id, org_id.as_deref(), json).await + } + ClickPipeCommands::Get { + service_id, + clickpipe_id, + org_id, + } => clickpipe_get(client, &service_id, &clickpipe_id, org_id.as_deref(), json).await, + ClickPipeCommands::Delete { + service_id, + clickpipe_id, + org_id, + } => clickpipe_delete(client, &service_id, &clickpipe_id, org_id.as_deref(), json).await, + ClickPipeCommands::Start { + service_id, + clickpipe_id, + org_id, + } => { + clickpipe_state( + client, + &service_id, + &clickpipe_id, + "start", + org_id.as_deref(), + json, + ) + .await + } + ClickPipeCommands::Stop { + service_id, + clickpipe_id, + org_id, + } => { + clickpipe_state( + client, + &service_id, + &clickpipe_id, + "stop", + org_id.as_deref(), + json, + ) + .await + } + ClickPipeCommands::Resync { + service_id, + clickpipe_id, + org_id, + } => { + clickpipe_state( + client, + &service_id, + &clickpipe_id, + "resync", + org_id.as_deref(), + json, + ) + .await + } + ClickPipeCommands::Scale { + service_id, + clickpipe_id, + replicas, + cpu_millicores, + memory_gb, + org_id, + } => { + clickpipe_scale( + client, + &service_id, + &clickpipe_id, + replicas, + cpu_millicores, + memory_gb, + org_id.as_deref(), + json, + ) + .await + } + ClickPipeCommands::Settings { command } => match command { + ClickPipeSettingsCommands::Get { + service_id, + clickpipe_id, + org_id, + } => { + clickpipe_settings_get(client, &service_id, &clickpipe_id, org_id.as_deref(), json) + .await + } + ClickPipeSettingsCommands::Update { + service_id, + clickpipe_id, + streaming_max_insert_wait_ms, + object_storage_concurrency, + object_storage_polling_interval_ms, + object_storage_max_insert_bytes, + object_storage_max_file_count, + clickhouse_max_threads, + clickhouse_max_insert_threads, + object_storage_use_cluster_function, + clickhouse_parallel_view_processing, + org_id, + } => { + clickpipe_settings_update( + client, + &service_id, + &clickpipe_id, + streaming_max_insert_wait_ms, + object_storage_concurrency, + object_storage_polling_interval_ms, + object_storage_max_insert_bytes, + object_storage_max_file_count, + clickhouse_max_threads, + clickhouse_max_insert_threads, + object_storage_use_cluster_function, + clickhouse_parallel_view_processing, + org_id.as_deref(), + json, + ) + .await + } + }, + ClickPipeCommands::SchemaDiscover { + service_id, + command, + org_id, + } => { + clickpipe_schema_discover(client, &service_id, &command, org_id.as_deref(), json).await + } + ClickPipeCommands::Create { command } => match command { + ClickPipeCreateCommands::ObjectStorage(args) => { + clickpipe_create_object_storage(client, &args, json).await + } + ClickPipeCreateCommands::Kafka(args) => { + clickpipe_create_kafka(client, &args, json).await + } + ClickPipeCreateCommands::Kinesis(args) => { + clickpipe_create_kinesis(client, &args, json).await + } + ClickPipeCreateCommands::Postgres(args) => { + clickpipe_create_postgres(client, &args, json).await + } + ClickPipeCreateCommands::MySQL(args) => { + clickpipe_create_mysql(client, &args, json).await + } + ClickPipeCreateCommands::MongoDB(args) => { + clickpipe_create_mongodb(client, &args, json).await + } + ClickPipeCreateCommands::BigQuery(args) => { + clickpipe_create_bigquery(client, &args, json).await + } + }, + } +} + +async fn clickpipe_list( + client: &CloudClient, + service_id: &str, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + let org_id = resolve_org_id(client, org_id).await?; + let clickpipes = client.list_clickpipes(&org_id, service_id).await?; + + if json { + println!("{}", serde_json::to_string_pretty(&clickpipes)?); + } else if clickpipes.is_empty() { + println!("No ClickPipes found"); + } else { + println!("ClickPipes:"); + for clickpipe in &clickpipes { + println!( + " {} ({}) - {}", + or_absent(clickpipe.name.as_deref()), + or_absent(clickpipe.id.as_ref()), + or_absent(clickpipe.state.as_ref()) + ); + } + } + Ok(()) +} + +async fn clickpipe_create_object_storage( + client: &CloudClient, + args: &ObjectStorageCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipePostObjectStorageSource, ClickPipePostObjectStorageSourceAuthentication, + ClickPipePostRequest, ClickPipePostSource, MskIamUser, + }; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + let parsed_columns = parse_columns(&args.columns)?; + + let (authentication, iam_role_val, access_key) = match ( + args.iam_role.as_deref(), + args.access_key_id.as_deref(), + args.secret_key.as_deref(), + ) { + (Some(role), _, _) => ( + Some(ClickPipePostObjectStorageSourceAuthentication::IAM_ROLE), + Some(role.to_string()), + None, + ), + (_, Some(key_id), Some(secret)) => ( + Some(ClickPipePostObjectStorageSourceAuthentication::IAM_USER), + None, + Some(MskIamUser { + access_key_id: key_id.to_string(), + secret_key: secret.to_string(), + }), + ), + _ => (None, None, None), + }; + let authentication = authentication + .or_else(|| { + args.connection_string + .as_ref() + .map(|_| ClickPipePostObjectStorageSourceAuthentication::CONNECTION_STRING) + }) + .or_else(|| { + args.service_account_file + .as_ref() + .map(|_| ClickPipePostObjectStorageSourceAuthentication::SERVICE_ACCOUNT) + }); + + let service_account_key = match args.service_account_file.as_deref() { + Some(path) => Some(read_gcp_service_account_file(path)?), + None => None, + }; + + let source = ClickPipePostObjectStorageSource { + r#type: parse_enum(&args.storage_type)?, + format: parse_enum(&args.format)?, + url: args.source_url.clone(), + compression: Some(parse_enum(&args.compression)?), + is_continuous: if args.continuous { Some(true) } else { None }, + queue_url: args.queue_url.clone(), + delimiter: args.delimiter.clone(), + authentication, + iam_role: iam_role_val, + access_key, + connection_string: args.connection_string.clone(), + azure_container_name: args.azure_container_name.clone(), + path: args.path.clone(), + service_account_key, + skip_initial_load: if args.skip_initial_load { + Some(true) + } else { + None + }, + start_after: args.start_after.clone(), + }; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + object_storage: Some(source), + ..Default::default() + }, + destination: build_destination(&args.database, &args.table, parsed_columns), + ..Default::default() + }; + + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +/// Build the Kafka `credentials` JSON body, whose shape is a `oneOf` determined +/// by the auth mode (see the `ClickPipePostKafkaSource.credentials` schema). +/// IAM_ROLE sends a null body — the role ARN flows through the separate +/// top-level `iamRole` field on the source, not through credentials. +/// +/// `mtls_contents` is the pre-read (certificate, privateKey) PEM bundle used +/// only for MUTUAL_TLS; the caller reads these from disk so this function +/// stays pure and testable. +fn build_kafka_credentials( + authentication: &clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication, + args: &KafkaSourceFields, + mtls_contents: Option<(String, String)>, +) -> Result { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + match authentication { + Auth::PLAIN | Auth::SCRAM_SHA_256 | Auth::SCRAM_SHA_512 => { + match (args.username.as_deref(), args.password.as_deref()) { + (Some(username), Some(password)) => { + Ok(serde_json::json!({ "username": username, "password": password })) + } + _ => Err(format!( + "{} requires --username and --password", + args.auth.as_deref().unwrap_or("PLAIN") + )), + } + } + Auth::IAM_USER => match (args.access_key_id.as_deref(), args.secret_key.as_deref()) { + (Some(access_key_id), Some(secret_key)) => Ok(serde_json::json!({ + "accessKeyId": access_key_id, + "secretKey": secret_key + })), + _ => Err("IAM_USER requires --access-key-id and --secret-key".into()), + }, + Auth::IAM_ROLE => { + if args.iam_role.is_none() { + Err("IAM_ROLE requires --iam-role".into()) + } else { + Ok(serde_json::Value::Null) + } + } + Auth::MUTUAL_TLS => match mtls_contents { + Some((certificate, private_key)) => Ok(serde_json::json!({ + "certificate": certificate, + "privateKey": private_key + })), + None => Err("MUTUAL_TLS requires --client-certificate and --client-key".into()), + }, + Auth::Unknown(_) => Ok(serde_json::Value::Null), + } +} + +/// Build a `ClickPipePostKafkaSource` from the CLI args, performing all +/// authentication/credential/schema-registry/CA validation up front so bad +/// invocations fail fast before any network call. Shared by the +/// `clickpipe create kafka` and `clickpipe schema-discover kafka` +/// handlers. +fn build_kafka_source( + args: &KafkaSourceFields, +) -> Result> { + use clickhouse_cloud_api::models::{ + ClickPipeKafkaOffset, ClickPipeKafkaSchemaRegistryCredentials, + ClickPipeMutateKafkaSchemaRegistry, ClickPipePostKafkaSource, + ClickPipePostKafkaSourceAuthentication, + }; + + let authentication: ClickPipePostKafkaSourceAuthentication = match args.auth.as_deref() { + Some(authentication) => parse_enum(authentication)?, + None => ClickPipePostKafkaSourceAuthentication::default(), + }; + + let mtls_cert_contents = match ( + &authentication, + args.client_certificate.as_deref(), + args.client_key.as_deref(), + ) { + (ClickPipePostKafkaSourceAuthentication::MUTUAL_TLS, Some(cert_path), Some(key_path)) => { + Some(( + std::fs::read_to_string(cert_path)?, + std::fs::read_to_string(key_path)?, + )) + } + _ => None, + }; + let credentials = build_kafka_credentials(&authentication, args, mtls_cert_contents)?; + + let schema_registry = args + .schema_registry_url + .as_ref() + .map(|url| -> Result<_, Box> { + let credentials = match ( + args.schema_registry_username.as_deref(), + args.schema_registry_password.as_deref(), + ) { + (Some(username), Some(password)) => ClickPipeKafkaSchemaRegistryCredentials { + username: username.to_string(), + password: password.to_string(), + }, + _ => ClickPipeKafkaSchemaRegistryCredentials::default(), + }; + let ca_certificate = match args.schema_registry_ca_certificate.as_deref() { + Some(path) => Some(std::fs::read_to_string(path)?), + None => None, + }; + Ok(ClickPipeMutateKafkaSchemaRegistry { + url: url.clone(), + authentication: Default::default(), + credentials, + ca_certificate, + }) + }) + .transpose()?; + + let ca_certificate = match args.ca_certificate.as_deref() { + Some(path) => Some(std::fs::read_to_string(path)?), + None => None, + }; + + Ok(ClickPipePostKafkaSource { + r#type: parse_enum(&args.kafka_type)?, + format: parse_enum(&args.format)?, + brokers: args.brokers.clone(), + topics: args.topics.clone(), + consumer_group: args.consumer_group.clone(), + exactly_once: None, + authentication, + credentials, + iam_role: args.iam_role.clone(), + offset: Some(ClickPipeKafkaOffset { + strategy: parse_enum(&args.offset)?, + timestamp: args.offset_timestamp.clone(), + }), + schema_registry, + ca_certificate, + reverse_private_endpoint_ids: args.reverse_private_endpoint_ids.clone(), + }) +} + +/// Build a `ClickPipePostKinesisSource` from the CLI args. Shared by the +/// `clickpipe create kinesis` and `clickpipe schema-discover kinesis` +/// handlers. +fn build_kinesis_source( + args: &KinesisSourceFields, +) -> Result> { + use clickhouse_cloud_api::models::{ClickPipePostKinesisSource, MskIamUser}; + + let access_key = match (args.access_key_id.as_deref(), args.secret_key.as_deref()) { + (Some(access_key_id), Some(secret_key)) => Some(MskIamUser { + access_key_id: access_key_id.to_string(), + secret_key: secret_key.to_string(), + }), + _ => None, + }; + + Ok(ClickPipePostKinesisSource { + format: parse_enum(&args.format)?, + stream_name: args.stream_name.clone(), + region: args.region.clone(), + authentication: parse_enum(&args.auth)?, + iam_role: args.iam_role.clone(), + access_key, + use_enhanced_fan_out: if args.enhanced_fan_out { + Some(true) + } else { + None + }, + iterator_type: parse_enum(&args.iterator_type)?, + timestamp: args + .iterator_timestamp + .map(|timestamp| { + i64::try_from(timestamp) + .map_err(|_| format!("--iterator-timestamp {timestamp} is out of range")) + }) + .transpose()?, + }) +} + +async fn clickpipe_create_kafka( + client: &CloudClient, + args: &KafkaCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ClickPipePostRequest, ClickPipePostSource}; + + // Validate args and build the source before any network call so bad + // invocations fail fast. + let parsed_columns = parse_columns(&args.columns)?; + let source = build_kafka_source(&args.source)?; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + kafka: Some(source), + ..Default::default() + }, + destination: build_destination(&args.database, &args.table, parsed_columns), + ..Default::default() + }; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +async fn clickpipe_create_kinesis( + client: &CloudClient, + args: &KinesisCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ClickPipePostRequest, ClickPipePostSource}; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + let parsed_columns = parse_columns(&args.columns)?; + let source = build_kinesis_source(&args.source)?; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + kinesis: Some(source), + ..Default::default() + }, + destination: build_destination(&args.database, &args.table, parsed_columns), + ..Default::default() + }; + + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +/// Discover the inferred schema for a Kafka or Kinesis source without creating +/// a ClickPipe (Beta). Side-effect-free, but the API gateway rejects +/// OAuth/Bearer on this POST endpoint, so it is classified as a write command +/// and requires API key auth. +async fn clickpipe_schema_discover( + client: &CloudClient, + service_id: &str, + command: &ClickPipeSchemaDiscoverCommands, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipeSchemaDiscoveryRequest, ClickPipeSchemaDiscoverySource, + }; + + let source = match command { + ClickPipeSchemaDiscoverCommands::Kafka(args) => ClickPipeSchemaDiscoverySource { + kafka: Some(build_kafka_source(args)?), + kinesis: None, + }, + ClickPipeSchemaDiscoverCommands::Kinesis(args) => ClickPipeSchemaDiscoverySource { + kafka: None, + kinesis: Some(build_kinesis_source(args)?), + }, + }; + + let request = ClickPipeSchemaDiscoveryRequest { source }; + let org_id = resolve_org_id(client, org_id).await?; + let response = client + .click_pipe_schema_discovery(&org_id, service_id, &request) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&response)?); + } else { + #[derive(Tabled)] + struct Row { + #[tabled(rename = "Name")] + name: String, + #[tabled(rename = "Type")] + r#type: String, + #[tabled(rename = "Optional")] + optional: String, + } + let rows: Vec = response + .fields + .unwrap_or_default() + .into_iter() + .map(|field| Row { + name: or_absent(field.name), + r#type: or_absent(field.r#type), + optional: match field.optional { + Some(true) => "true".to_string(), + Some(false) => "false".to_string(), + None => "".to_string(), + }, + }) + .collect(); + if rows.is_empty() { + println!("No fields discovered"); + } else { + println!("{}", Table::new(rows).with(Style::markdown())); + } + } + Ok(()) +} + +async fn clickpipe_get( + client: &CloudClient, + service_id: &str, + clickpipe_id: &str, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + let org_id = resolve_org_id(client, org_id).await?; + let clickpipe = client + .get_clickpipe(&org_id, service_id, clickpipe_id) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&clickpipe)?); + } else { + print_human(&clickpipe)?; + } + Ok(()) +} + +async fn clickpipe_delete( + client: &CloudClient, + service_id: &str, + clickpipe_id: &str, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + let org_id = resolve_org_id(client, org_id).await?; + client + .delete_clickpipe(&org_id, service_id, clickpipe_id) + .await?; + + if json { + println!("{}", serde_json::json!({ "deleted": clickpipe_id })); + } else { + println!("ClickPipe {} deleted", clickpipe_id); + } + Ok(()) +} + +async fn clickpipe_state( + client: &CloudClient, + service_id: &str, + clickpipe_id: &str, + command: &str, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::ClickPipeStatePatchRequestCommand; + let command_value = match command { + "start" => ClickPipeStatePatchRequestCommand::Start, + "stop" => ClickPipeStatePatchRequestCommand::Stop, + "resync" => ClickPipeStatePatchRequestCommand::Resync, + other => return Err(format!("Unknown state command: {}", other).into()), + }; + let org_id = resolve_org_id(client, org_id).await?; + let clickpipe = client + .change_clickpipe_state(&org_id, service_id, clickpipe_id, command_value) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&clickpipe)?); + } else { + println!( + "ClickPipe {} {} (state: {})", + or_absent(clickpipe.name.as_deref()), + command, + or_absent(clickpipe.state.as_ref()) + ); + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn clickpipe_scale( + client: &CloudClient, + service_id: &str, + clickpipe_id: &str, + replicas: Option, + cpu_millicores: Option, + memory_gb: Option, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + let org_id = resolve_org_id(client, org_id).await?; + let request = clickhouse_cloud_api::models::ClickPipeScalingPatchRequest { + replicas: replicas.map(i64::from), + replica_cpu_millicores: cpu_millicores.map(i64::from), + replica_memory_gb: memory_gb, + #[cfg(feature = "deprecated-fields")] + concurrency: None, + }; + let clickpipe = client + .update_clickpipe_scaling(&org_id, service_id, clickpipe_id, &request) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&clickpipe)?); + } else { + let scaling = clickpipe.scaling.unwrap_or_default(); + println!( + "ClickPipe {} scaling updated", + or_absent(clickpipe.name.as_deref()) + ); + println!(" Replicas: {}", or_absent(scaling.replicas)); + println!(" CPU: {}m", or_absent(scaling.replica_cpu_millicores)); + println!(" Memory: {} GB", or_absent(scaling.replica_memory_gb)); + } + Ok(()) +} + +async fn clickpipe_settings_get( + client: &CloudClient, + service_id: &str, + clickpipe_id: &str, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + let org_id = resolve_org_id(client, org_id).await?; + let settings = client + .get_clickpipe_settings(&org_id, service_id, clickpipe_id) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&settings)?); + } else { + print_human(&settings)?; + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn clickpipe_settings_update( + client: &CloudClient, + service_id: &str, + clickpipe_id: &str, + streaming_max_insert_wait_ms: Option, + object_storage_concurrency: Option, + object_storage_polling_interval_ms: Option, + object_storage_max_insert_bytes: Option, + object_storage_max_file_count: Option, + clickhouse_max_threads: Option, + clickhouse_max_insert_threads: Option, + object_storage_use_cluster_function: Option, + clickhouse_parallel_view_processing: Option, + org_id: Option<&str>, + json: bool, +) -> Result<(), Box> { + let org_id = resolve_org_id(client, org_id).await?; + let request = clickhouse_cloud_api::models::ClickPipeSettingsPutRequest { + streaming_max_insert_wait_ms: streaming_max_insert_wait_ms.map(i64::from), + object_storage_concurrency: object_storage_concurrency.map(i64::from), + object_storage_polling_interval_ms: object_storage_polling_interval_ms.map(i64::from), + object_storage_max_insert_bytes: object_storage_max_insert_bytes.map(|value| value as i64), + object_storage_max_file_count: object_storage_max_file_count.map(i64::from), + clickhouse_max_threads: clickhouse_max_threads.map(i64::from), + clickhouse_max_insert_threads: clickhouse_max_insert_threads.map(i64::from), + object_storage_use_cluster_function, + clickhouse_parallel_view_processing, + clickhouse_max_download_threads: None, + clickhouse_min_insert_block_size_bytes: None, + clickhouse_parallel_distributed_insert_select: None, + }; + let settings = client + .update_clickpipe_settings(&org_id, service_id, clickpipe_id, &request) + .await?; + + if json { + println!("{}", serde_json::to_string_pretty(&settings)?); + } else { + println!("ClickPipe settings updated"); + let value = serde_json::to_value(&settings)?; + if let Some(object) = value.as_object() { + for (key, value) in object { + if !value.is_null() { + println!(" {}: {}", key, value); + } + } + } + } + Ok(()) +} + +/// Parse a CLI string into a library enum. Library enums have a +/// `#[serde(untagged)] Unknown(String)` variant so unknown inputs are +/// forwarded to the API (which returns the canonical validation error). +fn parse_enum(value: &str) -> Result { + serde_json::from_value(serde_json::Value::String(value.to_string())) + .map_err(|error| format!("invalid value '{}': {}", value, error)) +} + +/// Parse `name:type` column specifications into library destination columns. +fn parse_columns( + columns: &[String], +) -> Result, String> { + columns + .iter() + .map(|column| { + let (name, column_type) = column + .split_once(':') + .ok_or_else(|| format!("Invalid column format '{}': expected name:type", column))?; + Ok(clickhouse_cloud_api::models::ClickPipeDestinationColumn { + name: name.to_string(), + r#type: column_type.to_string(), + }) + }) + .collect() +} + +/// Build a managed-table destination with the default MergeTree engine. +fn build_destination( + database: &str, + table: &str, + columns: Vec, +) -> clickhouse_cloud_api::models::ClickPipeMutateDestination { + // Database pipes (Postgres/MySQL/BigQuery) carry the destination table on + // the per-mapping `targetTable` and reject any of {table, managedTable, + // tableDefinition, columns} at the top level. Detect that case via empty + // `table` and emit a destination with only `database` populated. + if table.is_empty() { + return clickhouse_cloud_api::models::ClickPipeMutateDestination { + database: database.to_string(), + ..Default::default() + }; + } + clickhouse_cloud_api::models::ClickPipeMutateDestination { + database: database.to_string(), + table: Some(table.to_string()), + columns, + managed_table: Some(true), + roles: None, + table_definition: Some( + clickhouse_cloud_api::models::ClickPipeDestinationTableDefinition::default(), + ), + } +} + +/// Read a GCP service-account JSON key file from disk and return the +/// base64-encoded contents. Used by both the object-storage and BigQuery +/// `create` handlers — the upstream API wants the encoded blob regardless +/// of which source it ends up on. +fn read_gcp_service_account_file(path: &str) -> Result> { + let contents = std::fs::read_to_string(path)?; + Ok(base64::Engine::encode( + &base64::engine::general_purpose::STANDARD, + contents.as_bytes(), + )) +} + +/// Print the standard "created" confirmation for any create_* handler. +fn print_created( + clickpipe: &clickhouse_cloud_api::models::ClickPipe, + json: bool, +) -> Result<(), Box> { + if json { + println!("{}", serde_json::to_string_pretty(clickpipe)?); + } else { + println!("ClickPipe created successfully!"); + println!(" Name: {}", or_absent(clickpipe.name.as_deref())); + println!(" ID: {}", or_absent(clickpipe.id.as_ref())); + println!(" State: {}", or_absent(clickpipe.state.as_ref())); + } + Ok(()) +} + +/// Parse `schema.table:target_table` mappings into (schema, table, target) tuples. +/// Source-specific handlers map these into their own TableMapping struct. +fn parse_db_table_mappings(mappings: &[String]) -> Result, String> { + mappings + .iter() + .map(|mapping| { + let (source, target) = mapping.split_once(':').ok_or_else(|| { + format!( + "Invalid table mapping '{}': expected schema.table:target_table", + mapping + ) + })?; + let (schema, table) = source + .split_once('.') + .ok_or_else(|| format!("Invalid source '{}': expected schema.table", source))?; + Ok((schema.to_string(), table.to_string(), target.to_string())) + }) + .collect() +} + +async fn clickpipe_create_postgres( + client: &CloudClient, + args: &PostgresCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipeMutatePostgresSource, ClickPipePostRequest, ClickPipePostSource, + ClickPipePostgresPipeSettings, ClickPipePostgresPipeTableMapping, PLAIN, + }; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + let mappings = parse_db_table_mappings(&args.table_mappings)?; + + let ca_certificate = match args.ca_certificate.as_deref() { + Some(path) => Some(std::fs::read_to_string(path)?), + None => None, + }; + + let table_mappings = mappings + .into_iter() + .map( + |(source_schema_name, source_table, target_table)| ClickPipePostgresPipeTableMapping { + source_schema_name, + source_table, + target_table, + ..Default::default() + }, + ) + .collect(); + + let source = ClickPipeMutatePostgresSource { + r#type: Some(parse_enum(&args.postgres_type)?), + credentials: PLAIN { + username: args.username.clone(), + password: args.password.clone(), + }, + host: args.host.clone(), + port: i64::from(args.port), + database: args.pg_database.clone(), + disable_tls: false, + skip_cert_verification: false, + authentication: parse_enum(&args.auth)?, + iam_role: args.iam_role.clone(), + tls_host: args.tls_host.clone(), + ca_certificate, + settings: ClickPipePostgresPipeSettings { + replication_mode: parse_enum(&args.replication_mode)?, + publication_name: args.publication_name.clone(), + replication_slot_name: args.replication_slot_name.clone(), + ..Default::default() + }, + table_mappings, + }; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + postgres: Some(source), + ..Default::default() + }, + destination: build_destination("default", "", vec![]), + ..Default::default() + }; + + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +async fn clickpipe_create_mysql( + client: &CloudClient, + args: &MySqlCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipeMutateMySQLSource, ClickPipeMySQLPipeSettings, ClickPipeMySQLPipeTableMapping, + ClickPipePostRequest, ClickPipePostSource, PLAIN, + }; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + let mappings = parse_db_table_mappings(&args.table_mappings)?; + + let ca_certificate = match args.ca_certificate.as_deref() { + Some(path) => Some(std::fs::read_to_string(path)?), + None => None, + }; + + let table_mappings = mappings + .into_iter() + .map( + |(source_schema_name, source_table, target_table)| ClickPipeMySQLPipeTableMapping { + source_schema_name, + source_table, + target_table, + ..Default::default() + }, + ) + .collect(); + + let source = ClickPipeMutateMySQLSource { + r#type: Some(parse_enum(&args.mysql_type)?), + credentials: Some(PLAIN { + username: args.username.clone(), + password: args.password.clone(), + }), + host: args.host.clone(), + port: i64::from(args.port), + authentication: Some(parse_enum(&args.auth)?), + iam_role: args.iam_role.clone(), + tls_host: args.tls_host.clone(), + ca_certificate, + disable_tls: if args.disable_tls { Some(true) } else { None }, + skip_cert_verification: if args.skip_cert_verification { + Some(true) + } else { + None + }, + server_id: args.server_id.map(|value| value as i64), + settings: ClickPipeMySQLPipeSettings { + replication_mode: parse_enum(&args.replication_mode)?, + replication_mechanism: Some(parse_enum(&args.replication_mechanism)?), + ..Default::default() + }, + table_mappings, + }; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + mysql: Some(source), + ..Default::default() + }, + destination: build_destination("default", "", vec![]), + ..Default::default() + }; + + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +async fn clickpipe_create_mongodb( + client: &CloudClient, + args: &MongoDbCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipeMongoDBPipeSettings, ClickPipeMongoDBPipeTableMapping, + ClickPipeMutateMongoDBSource, ClickPipePostRequest, ClickPipePostSource, PLAIN, + }; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + + // MongoDB uses `database.collection:target_table` format. + let table_mappings: Vec = args + .table_mappings + .iter() + .map(|mapping| { + let (source, target_table) = mapping.split_once(':').ok_or_else(|| { + format!( + "Invalid table mapping '{}': expected database.collection:target_table", + mapping + ) + })?; + let (source_database_name, source_collection) = + source.split_once('.').ok_or_else(|| { + format!("Invalid source '{}': expected database.collection", source) + })?; + Ok(ClickPipeMongoDBPipeTableMapping { + source_database_name: source_database_name.to_string(), + source_collection: source_collection.to_string(), + target_table: target_table.to_string(), + table_engine: None, + }) + }) + .collect::, String>>()?; + + let ca_certificate = match args.ca_certificate.as_deref() { + Some(path) => Some(std::fs::read_to_string(path)?), + None => None, + }; + + let source = ClickPipeMutateMongoDBSource { + credentials: Some(PLAIN { + username: args.username.clone(), + password: args.password.clone(), + }), + uri: args.uri.clone(), + read_preference: parse_enum(&args.read_preference)?, + tls_host: args.tls_host.clone(), + ca_certificate, + disable_tls: if args.disable_tls { Some(true) } else { None }, + skip_cert_verification: None, + settings: ClickPipeMongoDBPipeSettings { + replication_mode: parse_enum(&args.replication_mode)?, + ..Default::default() + }, + table_mappings, + }; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + mongodb: Some(source), + ..Default::default() + }, + destination: build_destination("default", "", vec![]), + ..Default::default() + }; + + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +async fn clickpipe_create_bigquery( + client: &CloudClient, + args: &BigQueryCreateArgs, + json: bool, +) -> Result<(), Box> { + use clickhouse_cloud_api::models::{ + ClickPipeBigQueryPipeSettings, ClickPipeBigQueryPipeTableMapping, + ClickPipeMutateBigQuerySource, ClickPipePostRequest, ClickPipePostSource, ServiceAccount, + }; + + let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; + let service_account_file = read_gcp_service_account_file(&args.service_account_file)?; + + // BigQuery uses `dataset.table:target_table` format. + let table_mappings: Vec = args + .table_mappings + .iter() + .map(|mapping| { + let (source, target_table) = mapping.split_once(':').ok_or_else(|| { + format!( + "Invalid table mapping '{}': expected dataset.table:target_table", + mapping + ) + })?; + let (source_dataset_name, source_table) = source + .split_once('.') + .ok_or_else(|| format!("Invalid source '{}': expected dataset.table", source))?; + Ok(ClickPipeBigQueryPipeTableMapping { + source_dataset_name: source_dataset_name.to_string(), + source_table: source_table.to_string(), + target_table: target_table.to_string(), + ..Default::default() + }) + }) + .collect::, String>>()?; + + let source = ClickPipeMutateBigQuerySource { + credentials: ServiceAccount { + service_account_file, + }, + snapshot_staging_path: args.staging_path.clone(), + settings: ClickPipeBigQueryPipeSettings { + replication_mode: parse_enum("snapshot")?, + ..Default::default() + }, + table_mappings, + }; + + let request = ClickPipePostRequest { + name: args.name.clone(), + source: ClickPipePostSource { + bigquery: Some(source), + ..Default::default() + }, + destination: build_destination("default", "", vec![]), + ..Default::default() + }; + + let clickpipe = client + .create_clickpipe(&org_id, &args.service_id, &request) + .await?; + print_created(&clickpipe, json)?; + Ok(()) +} + +impl CloudClient { + pub async fn list_clickpipes( + &self, + org_id: &str, + service_id: &str, + ) -> crate::cloud::client::Result> { + let response = self + .api() + .click_pipe_get_list(org_id, service_id) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn get_clickpipe( + &self, + org_id: &str, + service_id: &str, + clickpipe_id: &str, + ) -> crate::cloud::client::Result { + let response = self + .api() + .click_pipe_get(org_id, service_id, clickpipe_id) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn create_clickpipe( + &self, + org_id: &str, + service_id: &str, + request: &clickhouse_cloud_api::models::ClickPipePostRequest, + ) -> crate::cloud::client::Result { + let response = self + .api() + .click_pipe_create(org_id, service_id, request) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn delete_clickpipe( + &self, + org_id: &str, + service_id: &str, + clickpipe_id: &str, + ) -> crate::cloud::client::Result { + let response = self + .api() + .click_pipe_delete(org_id, service_id, clickpipe_id) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Ok(crate::cloud::types::DeleteResponse { + status: response.status, + request_id: response.request_id, + }) + } + + pub async fn change_clickpipe_state( + &self, + org_id: &str, + service_id: &str, + clickpipe_id: &str, + command: clickhouse_cloud_api::models::ClickPipeStatePatchRequestCommand, + ) -> crate::cloud::client::Result { + use clickhouse_cloud_api::models::ClickPipeStatePatchRequest; + let request = ClickPipeStatePatchRequest { + command: Some(command), + }; + let response = self + .api() + .click_pipe_state_update(org_id, service_id, clickpipe_id, &request) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn update_clickpipe_scaling( + &self, + org_id: &str, + service_id: &str, + clickpipe_id: &str, + request: &clickhouse_cloud_api::models::ClickPipeScalingPatchRequest, + ) -> crate::cloud::client::Result { + let response = self + .api() + .click_pipe_scaling_update(org_id, service_id, clickpipe_id, request) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn get_clickpipe_settings( + &self, + org_id: &str, + service_id: &str, + clickpipe_id: &str, + ) -> crate::cloud::client::Result { + let response = self + .api() + .click_pipe_settings_get(org_id, service_id, clickpipe_id) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn update_clickpipe_settings( + &self, + org_id: &str, + service_id: &str, + clickpipe_id: &str, + request: &clickhouse_cloud_api::models::ClickPipeSettingsPutRequest, + ) -> crate::cloud::client::Result { + let response = self + .api() + .click_pipe_settings_update(org_id, service_id, clickpipe_id, request) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } + + pub async fn click_pipe_schema_discovery( + &self, + org_id: &str, + service_id: &str, + request: &clickhouse_cloud_api::models::ClickPipeSchemaDiscoveryRequest, + ) -> crate::cloud::client::Result + { + let response = self + .api() + .click_pipe_schema_discovery(org_id, service_id, request) + .await + .map_err(|error| self.convert_error_for_organization(error, org_id))?; + Self::unwrap_response(response) + } +} + #[cfg(test)] mod tests { use super::*; @@ -2658,4 +3948,408 @@ mod tests { true, ); } + + #[test] + fn build_kinesis_source_rejects_out_of_range_iterator_timestamp() { + let args = KinesisSourceFields { + stream_name: "stream".to_string(), + region: "us-east-1".to_string(), + format: "JSONEachRow".to_string(), + auth: "IAM_ROLE".to_string(), + iam_role: None, + access_key_id: None, + secret_key: None, + iterator_type: "AT_TIMESTAMP".to_string(), + iterator_timestamp: Some(u64::MAX), + enhanced_fan_out: false, + }; + let error = build_kinesis_source(&args).unwrap_err(); + assert!( + error.to_string().contains("out of range"), + "error should mention the range: {}", + error + ); + + let args = KinesisSourceFields { + iterator_timestamp: Some(1_750_000_000), + ..args + }; + let source = build_kinesis_source(&args).unwrap(); + assert_eq!(source.timestamp, Some(1_750_000_000)); + } + + #[test] + fn parse_db_table_mappings_valid() { + let mappings = vec![ + "public.users:public_users".to_string(), + "schema1.orders:schema1_orders".to_string(), + ]; + let result = parse_db_table_mappings(&mappings).unwrap(); + assert_eq!(result.len(), 2); + assert_eq!( + result[0], + ("public".into(), "users".into(), "public_users".into()) + ); + assert_eq!( + result[1], + ("schema1".into(), "orders".into(), "schema1_orders".into()) + ); + } + + #[test] + fn parse_db_table_mappings_missing_colon() { + let mappings = vec!["public.users".to_string()]; + let result = parse_db_table_mappings(&mappings); + assert!(result.is_err()); + assert!( + result + .unwrap_err() + .contains("expected schema.table:target_table") + ); + } + + #[test] + fn parse_db_table_mappings_missing_dot() { + let mappings = vec!["users:target".to_string()]; + let result = parse_db_table_mappings(&mappings); + assert!(result.is_err()); + assert!(result.unwrap_err().contains("expected schema.table")); + } + + #[test] + fn parse_db_table_mappings_empty() { + let mappings: Vec = vec![]; + let result = parse_db_table_mappings(&mappings).unwrap(); + assert!(result.is_empty()); + } + + #[test] + fn parse_enum_known_variant() { + use clickhouse_cloud_api::models::ClickPipePostObjectStorageSourceFormat; + let format: ClickPipePostObjectStorageSourceFormat = parse_enum("JSONEachRow").unwrap(); + assert_eq!(format, ClickPipePostObjectStorageSourceFormat::JSONEachRow); + } + + #[test] + fn parse_enum_unknown_falls_through() { + // Unknown values map to the catch-all Unknown(String) variant — + // forwarded to the API which returns the canonical validation error. + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceType; + let kafka_type: ClickPipePostKafkaSourceType = parse_enum("not-a-real-type").unwrap(); + assert_eq!( + kafka_type, + ClickPipePostKafkaSourceType::Unknown("not-a-real-type".to_string()) + ); + } + + #[test] + fn parse_enum_preserves_rename_spellings() { + // Enums use `#[serde(rename = "s3")]` etc. — wire format is authoritative. + use clickhouse_cloud_api::models::{ + ClickPipePostKafkaSourceAuthentication, ClickPipePostObjectStorageSourceType, + }; + let source_type: ClickPipePostObjectStorageSourceType = parse_enum("s3").unwrap(); + assert_eq!(source_type, ClickPipePostObjectStorageSourceType::S3); + let authentication: ClickPipePostKafkaSourceAuthentication = + parse_enum("SCRAM-SHA-256").unwrap(); + assert_eq!( + authentication, + ClickPipePostKafkaSourceAuthentication::SCRAM_SHA_256 + ); + } + + #[test] + fn parse_columns_valid() { + let columns = vec!["id:Int64".to_string(), "name:String".to_string()]; + let parsed = parse_columns(&columns).unwrap(); + assert_eq!(parsed.len(), 2); + assert_eq!(parsed[0].name, "id"); + assert_eq!(parsed[0].r#type, "Int64"); + assert_eq!(parsed[1].name, "name"); + assert_eq!(parsed[1].r#type, "String"); + } + + #[test] + fn parse_columns_missing_colon_errors() { + let columns = vec!["id_without_type".to_string()]; + let error = parse_columns(&columns).unwrap_err(); + assert!(error.contains("expected name:type")); + } + + #[test] + fn build_destination_uses_defaults_for_table_definition() { + let destination = build_destination("mydb", "events", vec![]); + assert_eq!(destination.database, "mydb"); + assert_eq!(destination.table.as_deref(), Some("events")); + assert_eq!(destination.managed_table, Some(true)); + // Default table engine is MergeTree, not something else. + assert_eq!( + destination + .table_definition + .as_ref() + .expect("non-database pipe gets a tableDefinition") + .engine + .r#type, + clickhouse_cloud_api::models::ClickPipeDestinationTableEngineType::MergeTree + ); + } + + #[test] + fn build_destination_omits_table_fields_for_database_pipes() { + let destination = build_destination("default", "", vec![]); + assert_eq!(destination.database, "default"); + assert_eq!(destination.table, None); + assert!(destination.columns.is_empty()); + assert_eq!(destination.managed_table, None); + assert_eq!(destination.roles, None); + assert_eq!(destination.table_definition, None); + } + + // `build_kafka_credentials` tests — lock the wire shape for each auth mode. + // Authoritative source: `ClickPipePostKafkaSource.credentials` in + // `crates/clickhouse-cloud-api/clickhouse_cloud_openapi.json`. + + fn kafka_args() -> KafkaCreateArgs { + KafkaCreateArgs { + service_id: "svc".into(), + name: "pipe".into(), + source: KafkaSourceFields { + brokers: "b:9092".into(), + topics: "t".into(), + format: "JSONEachRow".into(), + kafka_type: "kafka".into(), + consumer_group: None, + auth: None, + username: None, + password: None, + iam_role: None, + access_key_id: None, + secret_key: None, + offset: "from_beginning".into(), + offset_timestamp: None, + schema_registry_url: None, + schema_registry_username: None, + schema_registry_password: None, + ca_certificate: None, + client_certificate: None, + client_key: None, + schema_registry_ca_certificate: None, + reverse_private_endpoint_ids: vec![], + }, + database: "d".into(), + table: "t".into(), + columns: vec![], + org_id: None, + } + } + + #[test] + fn build_kafka_source_supports_minimal_fields() { + let mut args = kafka_args().source; + args.auth = Some("PLAIN".into()); + args.username = Some("user".into()); + args.password = Some("password".into()); + + let source = build_kafka_source(&args).unwrap(); + assert_eq!(source.r#type.to_string(), "kafka"); + assert_eq!(source.format.to_string(), "JSONEachRow"); + assert_eq!(source.brokers, "b:9092"); + assert_eq!(source.topics, "t"); + assert_eq!(source.authentication.to_string(), "PLAIN"); + assert_eq!(source.credentials["username"], "user"); + assert_eq!(source.credentials["password"], "password"); + assert_eq!(source.consumer_group, None); + assert_eq!(source.exactly_once, None); + assert_eq!(source.iam_role, None); + assert_eq!(source.ca_certificate, None); + assert_eq!(source.schema_registry, None); + assert!(source.reverse_private_endpoint_ids.is_empty()); + let offset = source.offset.expect("Kafka offset is always populated"); + assert_eq!(offset.strategy.to_string(), "from_beginning"); + assert_eq!(offset.timestamp, None); + } + + #[test] + fn build_kafka_source_supports_maximal_fields_and_certificate_files() { + let directory = tempfile::tempdir().unwrap(); + let broker_ca = directory.path().join("broker-ca.pem"); + let client_certificate = directory.path().join("client.pem"); + let client_key = directory.path().join("client.key"); + let registry_ca = directory.path().join("registry-ca.pem"); + std::fs::write(&broker_ca, "BROKER_CA").unwrap(); + std::fs::write(&client_certificate, "CLIENT_CERT").unwrap(); + std::fs::write(&client_key, "CLIENT_KEY").unwrap(); + std::fs::write(®istry_ca, "REGISTRY_CA").unwrap(); + + let mut args = kafka_args().source; + args.brokers = "broker-1:9092,broker-2:9092".into(); + args.topics = "topic-1,topic-2".into(); + args.format = "AvroConfluent".into(); + args.kafka_type = "msk".into(); + args.consumer_group = Some("group".into()); + args.auth = Some("MUTUAL_TLS".into()); + args.username = Some("user".into()); + args.password = Some("password".into()); + args.iam_role = Some("arn:role".into()); + args.access_key_id = Some("access".into()); + args.secret_key = Some("secret".into()); + args.offset = "from_timestamp".into(); + args.offset_timestamp = Some("2021-01-01T00:00".into()); + args.schema_registry_url = Some("https://registry.example".into()); + args.schema_registry_username = Some("registry-user".into()); + args.schema_registry_password = Some("registry-password".into()); + args.ca_certificate = Some(broker_ca.to_string_lossy().into_owned()); + args.client_certificate = Some(client_certificate.to_string_lossy().into_owned()); + args.client_key = Some(client_key.to_string_lossy().into_owned()); + args.schema_registry_ca_certificate = Some(registry_ca.to_string_lossy().into_owned()); + args.reverse_private_endpoint_ids = vec!["endpoint-1".into(), "endpoint-2".into()]; + + let source = build_kafka_source(&args).unwrap(); + assert_eq!(source.r#type.to_string(), "msk"); + assert_eq!(source.format.to_string(), "AvroConfluent"); + assert_eq!(source.brokers, "broker-1:9092,broker-2:9092"); + assert_eq!(source.topics, "topic-1,topic-2"); + assert_eq!(source.consumer_group.as_deref(), Some("group")); + assert_eq!(source.authentication.to_string(), "MUTUAL_TLS"); + assert_eq!(source.credentials["certificate"], "CLIENT_CERT"); + assert_eq!(source.credentials["privateKey"], "CLIENT_KEY"); + assert_eq!(source.iam_role.as_deref(), Some("arn:role")); + assert_eq!(source.ca_certificate.as_deref(), Some("BROKER_CA")); + assert_eq!( + source.reverse_private_endpoint_ids, + ["endpoint-1", "endpoint-2"] + ); + let offset = source.offset.expect("Kafka offset is always populated"); + assert_eq!(offset.strategy.to_string(), "from_timestamp"); + assert_eq!(offset.timestamp.as_deref(), Some("2021-01-01T00:00")); + let registry = source + .schema_registry + .expect("schema registry is populated"); + assert_eq!(registry.url, "https://registry.example"); + assert_eq!(registry.credentials.username, "registry-user"); + assert_eq!(registry.credentials.password, "registry-password"); + assert_eq!(registry.ca_certificate.as_deref(), Some("REGISTRY_CA")); + } + + #[test] + fn build_kinesis_source_supports_minimal_fields() { + let args = KinesisSourceFields { + stream_name: "stream".into(), + region: "us-east-1".into(), + format: "JSONEachRow".into(), + auth: "IAM_ROLE".into(), + iam_role: None, + access_key_id: None, + secret_key: None, + iterator_type: "TRIM_HORIZON".into(), + iterator_timestamp: None, + enhanced_fan_out: false, + }; + + let source = build_kinesis_source(&args).unwrap(); + assert_eq!(source.stream_name, "stream"); + assert_eq!(source.region, "us-east-1"); + assert_eq!(source.format.to_string(), "JSONEachRow"); + assert_eq!(source.authentication.to_string(), "IAM_ROLE"); + assert_eq!(source.iterator_type.to_string(), "TRIM_HORIZON"); + assert_eq!(source.iam_role, None); + assert_eq!(source.access_key, None); + assert_eq!(source.timestamp, None); + assert_eq!(source.use_enhanced_fan_out, None); + } + + #[test] + fn build_kinesis_source_supports_maximal_fields() { + let args = KinesisSourceFields { + stream_name: "stream".into(), + region: "us-east-1".into(), + format: "AvroConfluent".into(), + auth: "IAM_USER".into(), + iam_role: Some("arn:role".into()), + access_key_id: Some("access".into()), + secret_key: Some("secret".into()), + iterator_type: "AT_TIMESTAMP".into(), + iterator_timestamp: Some(1_750_000_000), + enhanced_fan_out: true, + }; + + let source = build_kinesis_source(&args).unwrap(); + assert_eq!(source.stream_name, "stream"); + assert_eq!(source.region, "us-east-1"); + assert_eq!(source.format.to_string(), "AvroConfluent"); + assert_eq!(source.authentication.to_string(), "IAM_USER"); + assert_eq!(source.iterator_type.to_string(), "AT_TIMESTAMP"); + assert_eq!(source.iam_role.as_deref(), Some("arn:role")); + let access_key = source.access_key.expect("access key is populated"); + assert_eq!(access_key.access_key_id, "access"); + assert_eq!(access_key.secret_key, "secret"); + assert_eq!(source.timestamp, Some(1_750_000_000)); + assert_eq!(source.use_enhanced_fan_out, Some(true)); + } + + #[test] + fn kafka_credentials_plain_shape() { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + let mut args = kafka_args(); + args.source.auth = Some("PLAIN".into()); + args.source.username = Some("u".into()); + args.source.password = Some("p".into()); + let credentials = build_kafka_credentials(&Auth::PLAIN, &args.source, None).unwrap(); + assert_eq!(credentials["username"], "u"); + assert_eq!(credentials["password"], "p"); + } + + #[test] + fn kafka_credentials_iam_user_shape() { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + let mut args = kafka_args(); + args.source.auth = Some("IAM_USER".into()); + args.source.access_key_id = Some("AKIA".into()); + args.source.secret_key = Some("secret".into()); + let credentials = build_kafka_credentials(&Auth::IAM_USER, &args.source, None).unwrap(); + // MskIamUser wire shape is {accessKeyId, secretKey} — NOT snake_case. + assert_eq!(credentials["accessKeyId"], "AKIA"); + assert_eq!(credentials["secretKey"], "secret"); + assert!(credentials.get("access_key_id").is_none()); + } + + #[test] + fn kafka_credentials_iam_role_is_null() { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + let mut args = kafka_args(); + args.source.auth = Some("IAM_ROLE".into()); + args.source.iam_role = Some("arn:aws:iam::123:role/Foo".into()); + // IAM_ROLE sends credentials=null; the role ARN flows through the + // top-level `iamRole` field on the Kafka source, not credentials. + let credentials = build_kafka_credentials(&Auth::IAM_ROLE, &args.source, None).unwrap(); + assert!(credentials.is_null()); + } + + #[test] + fn kafka_credentials_mutual_tls_shape() { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + let args = kafka_args(); + let contents = Some(("CERT_PEM".into(), "KEY_PEM".into())); + let credentials = + build_kafka_credentials(&Auth::MUTUAL_TLS, &args.source, contents).unwrap(); + assert_eq!(credentials["certificate"], "CERT_PEM"); + assert_eq!(credentials["privateKey"], "KEY_PEM"); + } + + #[test] + fn kafka_credentials_iam_user_missing_args_errors() { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + let args = kafka_args(); + let error = build_kafka_credentials(&Auth::IAM_USER, &args.source, None).unwrap_err(); + assert!(error.contains("--access-key-id")); + } + + #[test] + fn kafka_credentials_iam_role_missing_arn_errors() { + use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; + let mut args = kafka_args(); + args.source.auth = Some("IAM_ROLE".into()); + let error = build_kafka_credentials(&Auth::IAM_ROLE, &args.source, None).unwrap_err(); + assert!(error.contains("--iam-role")); + } } diff --git a/crates/clickhousectl/src/cloud/client.rs b/crates/clickhousectl/src/cloud/client.rs index 036eba53..f832f8bb 100644 --- a/crates/clickhousectl/src/cloud/client.rs +++ b/crates/clickhousectl/src/cloud/client.rs @@ -1,4 +1,3 @@ -use crate::cloud::types::DeleteResponse; use crate::dotenv::DotenvVars; use std::env; @@ -387,7 +386,7 @@ impl CloudClient { &self.base_url } - /// Access the library client for migrated commands. + /// Access the library client from domain-specific wrapper methods. pub fn api(&self) -> &clickhouse_cloud_api::Client { &self.lib_client } @@ -449,142 +448,6 @@ impl CloudClient { other => CloudError::new(other.to_string()), } } - - // ClickPipe endpoints (delegated to library client) - pub async fn list_clickpipes( - &self, - org_id: &str, - service_id: &str, - ) -> Result> { - let response = self - .api() - .click_pipe_get_list(org_id, service_id) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn get_clickpipe( - &self, - org_id: &str, - service_id: &str, - clickpipe_id: &str, - ) -> Result { - let response = self - .api() - .click_pipe_get(org_id, service_id, clickpipe_id) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn create_clickpipe( - &self, - org_id: &str, - service_id: &str, - request: &clickhouse_cloud_api::models::ClickPipePostRequest, - ) -> Result { - let response = self - .api() - .click_pipe_create(org_id, service_id, request) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn delete_clickpipe( - &self, - org_id: &str, - service_id: &str, - clickpipe_id: &str, - ) -> Result { - let response = self - .api() - .click_pipe_delete(org_id, service_id, clickpipe_id) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Ok(DeleteResponse { - status: response.status, - request_id: response.request_id, - }) - } - - pub async fn change_clickpipe_state( - &self, - org_id: &str, - service_id: &str, - clickpipe_id: &str, - command: clickhouse_cloud_api::models::ClickPipeStatePatchRequestCommand, - ) -> Result { - use clickhouse_cloud_api::models::ClickPipeStatePatchRequest; - let request = ClickPipeStatePatchRequest { - command: Some(command), - }; - let response = self - .api() - .click_pipe_state_update(org_id, service_id, clickpipe_id, &request) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn update_clickpipe_scaling( - &self, - org_id: &str, - service_id: &str, - clickpipe_id: &str, - request: &clickhouse_cloud_api::models::ClickPipeScalingPatchRequest, - ) -> Result { - let response = self - .api() - .click_pipe_scaling_update(org_id, service_id, clickpipe_id, request) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn get_clickpipe_settings( - &self, - org_id: &str, - service_id: &str, - clickpipe_id: &str, - ) -> Result { - let response = self - .api() - .click_pipe_settings_get(org_id, service_id, clickpipe_id) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn update_clickpipe_settings( - &self, - org_id: &str, - service_id: &str, - clickpipe_id: &str, - request: &clickhouse_cloud_api::models::ClickPipeSettingsPutRequest, - ) -> Result { - let response = self - .api() - .click_pipe_settings_update(org_id, service_id, clickpipe_id, request) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } - - pub async fn click_pipe_schema_discovery( - &self, - org_id: &str, - service_id: &str, - request: &clickhouse_cloud_api::models::ClickPipeSchemaDiscoveryRequest, - ) -> Result { - let response = self - .api() - .click_pipe_schema_discovery(org_id, service_id, request) - .await - .map_err(|e| self.convert_error_for_organization(e, org_id))?; - Self::unwrap_response(response) - } } #[cfg(test)] diff --git a/crates/clickhousectl/src/cloud/commands.rs b/crates/clickhousectl/src/cloud/commands.rs deleted file mode 100644 index 79c88a20..00000000 --- a/crates/clickhousectl/src/cloud/commands.rs +++ /dev/null @@ -1,1235 +0,0 @@ -use crate::cloud::client::CloudClient; -use crate::cloud::output::{or_absent, print_human}; -use crate::cloud::shared::resolve_org_id; -use tabled::{Table, Tabled, settings::Style}; - -pub async fn clickpipe_list( - client: &CloudClient, - service_id: &str, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - let org_id = resolve_org_id(client, org_id).await?; - let clickpipes = client.list_clickpipes(&org_id, service_id).await?; - - if json { - println!("{}", serde_json::to_string_pretty(&clickpipes)?); - } else if clickpipes.is_empty() { - println!("No ClickPipes found"); - } else { - println!("ClickPipes:"); - for cp in &clickpipes { - println!( - " {} ({}) - {}", - or_absent(cp.name.as_deref()), - or_absent(cp.id.as_ref()), - or_absent(cp.state.as_ref()) - ); - } - } - Ok(()) -} - -pub async fn clickpipe_create_s3( - client: &CloudClient, - args: &crate::cloud::cli::ObjectStorageCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipePostObjectStorageSource, ClickPipePostObjectStorageSourceAuthentication, - ClickPipePostRequest, ClickPipePostSource, MskIamUser, - }; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - let parsed_columns = parse_columns(&args.columns)?; - - let (authentication, iam_role_val, access_key) = match ( - args.iam_role.as_deref(), - args.access_key_id.as_deref(), - args.secret_key.as_deref(), - ) { - (Some(role), _, _) => ( - Some(ClickPipePostObjectStorageSourceAuthentication::IAM_ROLE), - Some(role.to_string()), - None, - ), - (_, Some(key_id), Some(secret)) => ( - Some(ClickPipePostObjectStorageSourceAuthentication::IAM_USER), - None, - Some(MskIamUser { - access_key_id: key_id.to_string(), - secret_key: secret.to_string(), - }), - ), - _ => (None, None, None), - }; - let authentication = authentication - .or_else(|| { - args.connection_string - .as_ref() - .map(|_| ClickPipePostObjectStorageSourceAuthentication::CONNECTION_STRING) - }) - .or_else(|| { - args.service_account_file - .as_ref() - .map(|_| ClickPipePostObjectStorageSourceAuthentication::SERVICE_ACCOUNT) - }); - - let service_account_key = match args.service_account_file.as_deref() { - Some(path) => Some(read_gcp_service_account_file(path)?), - None => None, - }; - - let source = ClickPipePostObjectStorageSource { - r#type: parse_enum(&args.storage_type)?, - format: parse_enum(&args.format)?, - url: args.source_url.clone(), - compression: Some(parse_enum(&args.compression)?), - is_continuous: if args.continuous { Some(true) } else { None }, - queue_url: args.queue_url.clone(), - delimiter: args.delimiter.clone(), - authentication, - iam_role: iam_role_val, - access_key, - connection_string: args.connection_string.clone(), - azure_container_name: args.azure_container_name.clone(), - path: args.path.clone(), - service_account_key, - skip_initial_load: if args.skip_initial_load { - Some(true) - } else { - None - }, - start_after: args.start_after.clone(), - }; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - object_storage: Some(source), - ..Default::default() - }, - destination: build_destination(&args.database, &args.table, parsed_columns), - ..Default::default() - }; - - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -/// Build the Kafka `credentials` JSON body, whose shape is a `oneOf` determined -/// by the auth mode (see the `ClickPipePostKafkaSource.credentials` schema). -/// IAM_ROLE sends a null body — the role ARN flows through the separate -/// top-level `iamRole` field on the source, not through credentials. -/// -/// `mtls_contents` is the pre-read (certificate, privateKey) PEM bundle used -/// only for MUTUAL_TLS; the caller reads these from disk so this function -/// stays pure and testable. -fn build_kafka_credentials( - authentication: &clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication, - args: &crate::cloud::cli::KafkaSourceFields, - mtls_contents: Option<(String, String)>, -) -> Result { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - match authentication { - Auth::PLAIN | Auth::SCRAM_SHA_256 | Auth::SCRAM_SHA_512 => { - match (args.username.as_deref(), args.password.as_deref()) { - (Some(u), Some(p)) => Ok(serde_json::json!({ "username": u, "password": p })), - _ => Err(format!( - "{} requires --username and --password", - args.auth.as_deref().unwrap_or("PLAIN") - )), - } - } - Auth::IAM_USER => match (args.access_key_id.as_deref(), args.secret_key.as_deref()) { - (Some(k), Some(s)) => Ok(serde_json::json!({ "accessKeyId": k, "secretKey": s })), - _ => Err("IAM_USER requires --access-key-id and --secret-key".into()), - }, - Auth::IAM_ROLE => { - if args.iam_role.is_none() { - Err("IAM_ROLE requires --iam-role".into()) - } else { - Ok(serde_json::Value::Null) - } - } - Auth::MUTUAL_TLS => match mtls_contents { - Some((cert, key)) => Ok(serde_json::json!({ "certificate": cert, "privateKey": key })), - None => Err("MUTUAL_TLS requires --client-certificate and --client-key".into()), - }, - Auth::Unknown(_) => Ok(serde_json::Value::Null), - } -} - -/// Build a `ClickPipePostKafkaSource` from the CLI args, performing all -/// authentication/credential/schema-registry/CA validation up front so bad -/// invocations fail fast before any network call. Shared by the -/// `clickpipe create kafka` and `clickpipe schema-discover kafka` -/// handlers. -fn build_kafka_source( - args: &crate::cloud::cli::KafkaSourceFields, -) -> Result> { - use clickhouse_cloud_api::models::{ - ClickPipeKafkaOffset, ClickPipeKafkaSchemaRegistryCredentials, - ClickPipeMutateKafkaSchemaRegistry, ClickPipePostKafkaSource, - ClickPipePostKafkaSourceAuthentication, - }; - - let authentication: ClickPipePostKafkaSourceAuthentication = match args.auth.as_deref() { - Some(a) => parse_enum(a)?, - None => ClickPipePostKafkaSourceAuthentication::default(), - }; - - let mtls_cert_contents = match ( - &authentication, - args.client_certificate.as_deref(), - args.client_key.as_deref(), - ) { - (ClickPipePostKafkaSourceAuthentication::MUTUAL_TLS, Some(cert_path), Some(key_path)) => { - Some(( - std::fs::read_to_string(cert_path)?, - std::fs::read_to_string(key_path)?, - )) - } - _ => None, - }; - let credentials = build_kafka_credentials(&authentication, args, mtls_cert_contents)?; - - let schema_registry = args - .schema_registry_url - .as_ref() - .map(|url| -> Result<_, Box> { - let creds = match ( - args.schema_registry_username.as_deref(), - args.schema_registry_password.as_deref(), - ) { - (Some(u), Some(p)) => ClickPipeKafkaSchemaRegistryCredentials { - username: u.to_string(), - password: p.to_string(), - }, - _ => ClickPipeKafkaSchemaRegistryCredentials::default(), - }; - let ca_cert = match args.schema_registry_ca_certificate.as_deref() { - Some(path) => Some(std::fs::read_to_string(path)?), - None => None, - }; - Ok(ClickPipeMutateKafkaSchemaRegistry { - url: url.clone(), - authentication: Default::default(), - credentials: creds, - ca_certificate: ca_cert, - }) - }) - .transpose()?; - - let ca_cert_contents = match args.ca_certificate.as_deref() { - Some(path) => Some(std::fs::read_to_string(path)?), - None => None, - }; - - Ok(ClickPipePostKafkaSource { - r#type: parse_enum(&args.kafka_type)?, - format: parse_enum(&args.format)?, - brokers: args.brokers.clone(), - topics: args.topics.clone(), - consumer_group: args.consumer_group.clone(), - exactly_once: None, - authentication, - credentials, - iam_role: args.iam_role.clone(), - offset: Some(ClickPipeKafkaOffset { - strategy: parse_enum(&args.offset)?, - timestamp: args.offset_timestamp.clone(), - }), - schema_registry, - ca_certificate: ca_cert_contents, - reverse_private_endpoint_ids: args.reverse_private_endpoint_ids.clone(), - }) -} - -/// Build a `ClickPipePostKinesisSource` from the CLI args. Shared by the -/// `clickpipe create kinesis` and `clickpipe schema-discover kinesis` -/// handlers. -fn build_kinesis_source( - args: &crate::cloud::cli::KinesisSourceFields, -) -> Result> { - use clickhouse_cloud_api::models::{ClickPipePostKinesisSource, MskIamUser}; - - let access_key = match (args.access_key_id.as_deref(), args.secret_key.as_deref()) { - (Some(k), Some(s)) => Some(MskIamUser { - access_key_id: k.to_string(), - secret_key: s.to_string(), - }), - _ => None, - }; - - Ok(ClickPipePostKinesisSource { - format: parse_enum(&args.format)?, - stream_name: args.stream_name.clone(), - region: args.region.clone(), - authentication: parse_enum(&args.auth)?, - iam_role: args.iam_role.clone(), - access_key, - use_enhanced_fan_out: if args.enhanced_fan_out { - Some(true) - } else { - None - }, - iterator_type: parse_enum(&args.iterator_type)?, - timestamp: args - .iterator_timestamp - .map(|t| { - i64::try_from(t).map_err(|_| format!("--iterator-timestamp {t} is out of range")) - }) - .transpose()?, - }) -} - -pub async fn clickpipe_create_kafka( - client: &CloudClient, - args: &crate::cloud::cli::KafkaCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ClickPipePostRequest, ClickPipePostSource}; - - // Validate args and build the source before any network call so bad - // invocations fail fast. - let parsed_columns = parse_columns(&args.columns)?; - let source = build_kafka_source(&args.source)?; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - kafka: Some(source), - ..Default::default() - }, - destination: build_destination(&args.database, &args.table, parsed_columns), - ..Default::default() - }; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -pub async fn clickpipe_create_kinesis( - client: &CloudClient, - args: &crate::cloud::cli::KinesisCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ClickPipePostRequest, ClickPipePostSource}; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - let parsed_columns = parse_columns(&args.columns)?; - let source = build_kinesis_source(&args.source)?; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - kinesis: Some(source), - ..Default::default() - }, - destination: build_destination(&args.database, &args.table, parsed_columns), - ..Default::default() - }; - - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -/// Discover the inferred schema for a Kafka or Kinesis source without creating -/// a ClickPipe (Beta). Side-effect-free, but the API gateway rejects -/// OAuth/Bearer on this POST endpoint, so it is classified as a write command -/// and requires API key auth. -pub async fn clickpipe_schema_discover( - client: &CloudClient, - service_id: &str, - command: &crate::cloud::cli::ClickPipeSchemaDiscoverCommands, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipeSchemaDiscoveryRequest, ClickPipeSchemaDiscoverySource, - }; - - let source = match command { - crate::cloud::cli::ClickPipeSchemaDiscoverCommands::Kafka(args) => { - ClickPipeSchemaDiscoverySource { - kafka: Some(build_kafka_source(args)?), - kinesis: None, - } - } - crate::cloud::cli::ClickPipeSchemaDiscoverCommands::Kinesis(args) => { - ClickPipeSchemaDiscoverySource { - kafka: None, - kinesis: Some(build_kinesis_source(args)?), - } - } - }; - - let request = ClickPipeSchemaDiscoveryRequest { source }; - let org_id = resolve_org_id(client, org_id).await?; - let response = client - .click_pipe_schema_discovery(&org_id, service_id, &request) - .await?; - - if json { - println!("{}", serde_json::to_string_pretty(&response)?); - } else { - #[derive(Tabled)] - struct Row { - #[tabled(rename = "Name")] - name: String, - #[tabled(rename = "Type")] - r#type: String, - #[tabled(rename = "Optional")] - optional: String, - } - let rows: Vec = response - .fields - .unwrap_or_default() - .into_iter() - .map(|f| Row { - name: or_absent(f.name), - r#type: or_absent(f.r#type), - optional: match f.optional { - Some(true) => "true".to_string(), - Some(false) => "false".to_string(), - None => "".to_string(), - }, - }) - .collect(); - if rows.is_empty() { - println!("No fields discovered"); - } else { - println!("{}", Table::new(rows).with(Style::markdown())); - } - } - Ok(()) -} - -pub async fn clickpipe_get( - client: &CloudClient, - service_id: &str, - clickpipe_id: &str, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - let org_id = resolve_org_id(client, org_id).await?; - let clickpipe = client - .get_clickpipe(&org_id, service_id, clickpipe_id) - .await?; - - if json { - println!("{}", serde_json::to_string_pretty(&clickpipe)?); - } else { - print_human(&clickpipe)?; - } - Ok(()) -} - -pub async fn clickpipe_delete( - client: &CloudClient, - service_id: &str, - clickpipe_id: &str, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - let org_id = resolve_org_id(client, org_id).await?; - client - .delete_clickpipe(&org_id, service_id, clickpipe_id) - .await?; - - if json { - println!("{}", serde_json::json!({ "deleted": clickpipe_id })); - } else { - println!("ClickPipe {} deleted", clickpipe_id); - } - Ok(()) -} - -pub async fn clickpipe_state( - client: &CloudClient, - service_id: &str, - clickpipe_id: &str, - command: &str, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::ClickPipeStatePatchRequestCommand; - let cmd = match command { - "start" => ClickPipeStatePatchRequestCommand::Start, - "stop" => ClickPipeStatePatchRequestCommand::Stop, - "resync" => ClickPipeStatePatchRequestCommand::Resync, - other => return Err(format!("Unknown state command: {}", other).into()), - }; - let org_id = resolve_org_id(client, org_id).await?; - let clickpipe = client - .change_clickpipe_state(&org_id, service_id, clickpipe_id, cmd) - .await?; - - if json { - println!("{}", serde_json::to_string_pretty(&clickpipe)?); - } else { - println!( - "ClickPipe {} {} (state: {})", - or_absent(clickpipe.name.as_deref()), - command, - or_absent(clickpipe.state.as_ref()) - ); - } - Ok(()) -} - -#[allow(clippy::too_many_arguments)] -pub async fn clickpipe_scale( - client: &CloudClient, - service_id: &str, - clickpipe_id: &str, - replicas: Option, - cpu_millicores: Option, - memory_gb: Option, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - let org_id = resolve_org_id(client, org_id).await?; - let request = clickhouse_cloud_api::models::ClickPipeScalingPatchRequest { - replicas: replicas.map(i64::from), - replica_cpu_millicores: cpu_millicores.map(i64::from), - replica_memory_gb: memory_gb, - #[cfg(feature = "deprecated-fields")] - concurrency: None, - }; - let clickpipe = client - .update_clickpipe_scaling(&org_id, service_id, clickpipe_id, &request) - .await?; - - if json { - println!("{}", serde_json::to_string_pretty(&clickpipe)?); - } else { - let scaling = clickpipe.scaling.unwrap_or_default(); - println!( - "ClickPipe {} scaling updated", - or_absent(clickpipe.name.as_deref()) - ); - println!(" Replicas: {}", or_absent(scaling.replicas)); - println!(" CPU: {}m", or_absent(scaling.replica_cpu_millicores)); - println!(" Memory: {} GB", or_absent(scaling.replica_memory_gb)); - } - Ok(()) -} - -pub async fn clickpipe_settings_get( - client: &CloudClient, - service_id: &str, - clickpipe_id: &str, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - let org_id = resolve_org_id(client, org_id).await?; - let settings = client - .get_clickpipe_settings(&org_id, service_id, clickpipe_id) - .await?; - - if json { - println!("{}", serde_json::to_string_pretty(&settings)?); - } else { - print_human(&settings)?; - } - Ok(()) -} - -#[allow(clippy::too_many_arguments)] -pub async fn clickpipe_settings_update( - client: &CloudClient, - service_id: &str, - clickpipe_id: &str, - streaming_max_insert_wait_ms: Option, - object_storage_concurrency: Option, - object_storage_polling_interval_ms: Option, - object_storage_max_insert_bytes: Option, - object_storage_max_file_count: Option, - clickhouse_max_threads: Option, - clickhouse_max_insert_threads: Option, - object_storage_use_cluster_function: Option, - clickhouse_parallel_view_processing: Option, - org_id: Option<&str>, - json: bool, -) -> Result<(), Box> { - let org_id = resolve_org_id(client, org_id).await?; - let request = clickhouse_cloud_api::models::ClickPipeSettingsPutRequest { - streaming_max_insert_wait_ms: streaming_max_insert_wait_ms.map(i64::from), - object_storage_concurrency: object_storage_concurrency.map(i64::from), - object_storage_polling_interval_ms: object_storage_polling_interval_ms.map(i64::from), - object_storage_max_insert_bytes: object_storage_max_insert_bytes.map(|v| v as i64), - object_storage_max_file_count: object_storage_max_file_count.map(i64::from), - clickhouse_max_threads: clickhouse_max_threads.map(i64::from), - clickhouse_max_insert_threads: clickhouse_max_insert_threads.map(i64::from), - object_storage_use_cluster_function, - clickhouse_parallel_view_processing, - clickhouse_max_download_threads: None, - clickhouse_min_insert_block_size_bytes: None, - clickhouse_parallel_distributed_insert_select: None, - }; - let settings = client - .update_clickpipe_settings(&org_id, service_id, clickpipe_id, &request) - .await?; - - if json { - println!("{}", serde_json::to_string_pretty(&settings)?); - } else { - println!("ClickPipe settings updated"); - let value = serde_json::to_value(&settings)?; - if let Some(obj) = value.as_object() { - for (key, val) in obj { - if !val.is_null() { - println!(" {}: {}", key, val); - } - } - } - } - Ok(()) -} - -/// Parse a CLI string into a library enum. Library enums have a -/// `#[serde(untagged)] Unknown(String)` variant so unknown inputs are -/// forwarded to the API (which returns the canonical validation error). -fn parse_enum(s: &str) -> Result { - serde_json::from_value(serde_json::Value::String(s.to_string())) - .map_err(|e| format!("invalid value '{}': {}", s, e)) -} - -/// Parse `name:type` column specifications into library destination columns. -fn parse_columns( - columns: &[String], -) -> Result, String> { - columns - .iter() - .map(|col| { - let (name, col_type) = col - .split_once(':') - .ok_or_else(|| format!("Invalid column format '{}': expected name:type", col))?; - Ok(clickhouse_cloud_api::models::ClickPipeDestinationColumn { - name: name.to_string(), - r#type: col_type.to_string(), - }) - }) - .collect() -} - -/// Build a managed-table destination with the default MergeTree engine. -fn build_destination( - database: &str, - table: &str, - columns: Vec, -) -> clickhouse_cloud_api::models::ClickPipeMutateDestination { - // Database pipes (Postgres/MySQL/BigQuery) carry the destination table on - // the per-mapping `targetTable` and reject any of {table, managedTable, - // tableDefinition, columns} at the top level. Detect that case via empty - // `table` and emit a destination with only `database` populated. - if table.is_empty() { - return clickhouse_cloud_api::models::ClickPipeMutateDestination { - database: database.to_string(), - ..Default::default() - }; - } - clickhouse_cloud_api::models::ClickPipeMutateDestination { - database: database.to_string(), - table: Some(table.to_string()), - columns, - managed_table: Some(true), - roles: None, - table_definition: Some( - clickhouse_cloud_api::models::ClickPipeDestinationTableDefinition::default(), - ), - } -} - -/// Read a GCP service-account JSON key file from disk and return the -/// base64-encoded contents. Used by both the object-storage and BigQuery -/// `create` handlers — the upstream API wants the encoded blob regardless -/// of which source it ends up on. -fn read_gcp_service_account_file(path: &str) -> Result> { - let contents = std::fs::read_to_string(path)?; - Ok(base64::Engine::encode( - &base64::engine::general_purpose::STANDARD, - contents.as_bytes(), - )) -} - -/// Print the standard "created" confirmation for any create_* handler. -fn print_created( - clickpipe: &clickhouse_cloud_api::models::ClickPipe, - json: bool, -) -> Result<(), Box> { - if json { - println!("{}", serde_json::to_string_pretty(clickpipe)?); - } else { - println!("ClickPipe created successfully!"); - println!(" Name: {}", or_absent(clickpipe.name.as_deref())); - println!(" ID: {}", or_absent(clickpipe.id.as_ref())); - println!(" State: {}", or_absent(clickpipe.state.as_ref())); - } - Ok(()) -} - -/// Parse `schema.table:target_table` mappings into (schema, table, target) tuples. -/// Source-specific handlers map these into their own TableMapping struct. -fn parse_db_table_mappings(mappings: &[String]) -> Result, String> { - mappings - .iter() - .map(|m| { - let (source, target) = m.split_once(':').ok_or_else(|| { - format!( - "Invalid table mapping '{}': expected schema.table:target_table", - m - ) - })?; - let (schema, table) = source - .split_once('.') - .ok_or_else(|| format!("Invalid source '{}': expected schema.table", source))?; - Ok((schema.to_string(), table.to_string(), target.to_string())) - }) - .collect() -} - -pub async fn clickpipe_create_postgres( - client: &CloudClient, - args: &crate::cloud::cli::PostgresCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipeMutatePostgresSource, ClickPipePostRequest, ClickPipePostSource, - ClickPipePostgresPipeSettings, ClickPipePostgresPipeTableMapping, PLAIN, - }; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - let mappings = parse_db_table_mappings(&args.table_mappings)?; - - let ca_cert_contents = match args.ca_certificate.as_deref() { - Some(path) => Some(std::fs::read_to_string(path)?), - None => None, - }; - - let pg_mappings = mappings - .into_iter() - .map(|(schema, t, target)| ClickPipePostgresPipeTableMapping { - source_schema_name: schema, - source_table: t, - target_table: target, - ..Default::default() - }) - .collect(); - - let source = ClickPipeMutatePostgresSource { - r#type: Some(parse_enum(&args.postgres_type)?), - credentials: PLAIN { - username: args.username.clone(), - password: args.password.clone(), - }, - host: args.host.clone(), - port: i64::from(args.port), - database: args.pg_database.clone(), - disable_tls: false, - skip_cert_verification: false, - authentication: parse_enum(&args.auth)?, - iam_role: args.iam_role.clone(), - tls_host: args.tls_host.clone(), - ca_certificate: ca_cert_contents, - settings: ClickPipePostgresPipeSettings { - replication_mode: parse_enum(&args.replication_mode)?, - publication_name: args.publication_name.clone(), - replication_slot_name: args.replication_slot_name.clone(), - ..Default::default() - }, - table_mappings: pg_mappings, - }; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - postgres: Some(source), - ..Default::default() - }, - destination: build_destination("default", "", vec![]), - ..Default::default() - }; - - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -pub async fn clickpipe_create_mysql( - client: &CloudClient, - args: &crate::cloud::cli::MySqlCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipeMutateMySQLSource, ClickPipeMySQLPipeSettings, ClickPipeMySQLPipeTableMapping, - ClickPipePostRequest, ClickPipePostSource, PLAIN, - }; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - let mappings = parse_db_table_mappings(&args.table_mappings)?; - - let ca_cert_contents = match args.ca_certificate.as_deref() { - Some(path) => Some(std::fs::read_to_string(path)?), - None => None, - }; - - let mysql_mappings = mappings - .into_iter() - .map(|(schema, t, target)| ClickPipeMySQLPipeTableMapping { - source_schema_name: schema, - source_table: t, - target_table: target, - ..Default::default() - }) - .collect(); - - let source = ClickPipeMutateMySQLSource { - r#type: Some(parse_enum(&args.mysql_type)?), - credentials: Some(PLAIN { - username: args.username.clone(), - password: args.password.clone(), - }), - host: args.host.clone(), - port: i64::from(args.port), - authentication: Some(parse_enum(&args.auth)?), - iam_role: args.iam_role.clone(), - tls_host: args.tls_host.clone(), - ca_certificate: ca_cert_contents, - disable_tls: if args.disable_tls { Some(true) } else { None }, - skip_cert_verification: if args.skip_cert_verification { - Some(true) - } else { - None - }, - server_id: args.server_id.map(|v| v as i64), - settings: ClickPipeMySQLPipeSettings { - replication_mode: parse_enum(&args.replication_mode)?, - replication_mechanism: Some(parse_enum(&args.replication_mechanism)?), - ..Default::default() - }, - table_mappings: mysql_mappings, - }; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - mysql: Some(source), - ..Default::default() - }, - destination: build_destination("default", "", vec![]), - ..Default::default() - }; - - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -pub async fn clickpipe_create_mongodb( - client: &CloudClient, - args: &crate::cloud::cli::MongoDbCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipeMongoDBPipeSettings, ClickPipeMongoDBPipeTableMapping, - ClickPipeMutateMongoDBSource, ClickPipePostRequest, ClickPipePostSource, PLAIN, - }; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - - // MongoDB uses `database.collection:target_table` format. - let mongo_mappings: Vec = args - .table_mappings - .iter() - .map(|m| { - let (source, target) = m.split_once(':').ok_or_else(|| { - format!( - "Invalid table mapping '{}': expected database.collection:target_table", - m - ) - })?; - let (db, collection) = source.split_once('.').ok_or_else(|| { - format!("Invalid source '{}': expected database.collection", source) - })?; - Ok(ClickPipeMongoDBPipeTableMapping { - source_database_name: db.to_string(), - source_collection: collection.to_string(), - target_table: target.to_string(), - table_engine: None, - }) - }) - .collect::, String>>()?; - - let ca_cert_contents = match args.ca_certificate.as_deref() { - Some(path) => Some(std::fs::read_to_string(path)?), - None => None, - }; - - let source = ClickPipeMutateMongoDBSource { - credentials: Some(PLAIN { - username: args.username.clone(), - password: args.password.clone(), - }), - uri: args.uri.clone(), - read_preference: parse_enum(&args.read_preference)?, - tls_host: args.tls_host.clone(), - ca_certificate: ca_cert_contents, - disable_tls: if args.disable_tls { Some(true) } else { None }, - skip_cert_verification: None, - settings: ClickPipeMongoDBPipeSettings { - replication_mode: parse_enum(&args.replication_mode)?, - ..Default::default() - }, - table_mappings: mongo_mappings, - }; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - mongodb: Some(source), - ..Default::default() - }, - destination: build_destination("default", "", vec![]), - ..Default::default() - }; - - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -pub async fn clickpipe_create_bigquery( - client: &CloudClient, - args: &crate::cloud::cli::BigQueryCreateArgs, - json: bool, -) -> Result<(), Box> { - use clickhouse_cloud_api::models::{ - ClickPipeBigQueryPipeSettings, ClickPipeBigQueryPipeTableMapping, - ClickPipeMutateBigQuerySource, ClickPipePostRequest, ClickPipePostSource, ServiceAccount, - }; - - let org_id = resolve_org_id(client, args.org_id.as_deref()).await?; - let sa_b64 = read_gcp_service_account_file(&args.service_account_file)?; - - // BigQuery uses `dataset.table:target_table` format. - let bq_mappings: Vec = args - .table_mappings - .iter() - .map(|m| { - let (source, target) = m.split_once(':').ok_or_else(|| { - format!( - "Invalid table mapping '{}': expected dataset.table:target_table", - m - ) - })?; - let (dataset, t) = source - .split_once('.') - .ok_or_else(|| format!("Invalid source '{}': expected dataset.table", source))?; - Ok(ClickPipeBigQueryPipeTableMapping { - source_dataset_name: dataset.to_string(), - source_table: t.to_string(), - target_table: target.to_string(), - ..Default::default() - }) - }) - .collect::, String>>()?; - - let source = ClickPipeMutateBigQuerySource { - credentials: ServiceAccount { - service_account_file: sa_b64, - }, - snapshot_staging_path: args.staging_path.clone(), - settings: ClickPipeBigQueryPipeSettings { - replication_mode: parse_enum("snapshot")?, - ..Default::default() - }, - table_mappings: bq_mappings, - }; - - let request = ClickPipePostRequest { - name: args.name.clone(), - source: ClickPipePostSource { - bigquery: Some(source), - ..Default::default() - }, - destination: build_destination("default", "", vec![]), - ..Default::default() - }; - - let clickpipe = client - .create_clickpipe(&org_id, &args.service_id, &request) - .await?; - print_created(&clickpipe, json)?; - Ok(()) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn build_kinesis_source_rejects_out_of_range_iterator_timestamp() { - let args = crate::cloud::cli::KinesisSourceFields { - stream_name: "stream".to_string(), - region: "us-east-1".to_string(), - format: "JSONEachRow".to_string(), - auth: "IAM_ROLE".to_string(), - iam_role: None, - access_key_id: None, - secret_key: None, - iterator_type: "AT_TIMESTAMP".to_string(), - iterator_timestamp: Some(u64::MAX), - enhanced_fan_out: false, - }; - let err = build_kinesis_source(&args).unwrap_err(); - assert!( - err.to_string().contains("out of range"), - "error should mention the range: {}", - err - ); - - let args = crate::cloud::cli::KinesisSourceFields { - iterator_timestamp: Some(1_750_000_000), - ..args - }; - let source = build_kinesis_source(&args).unwrap(); - assert_eq!(source.timestamp, Some(1_750_000_000)); - } - - #[test] - fn parse_db_table_mappings_valid() { - let mappings = vec![ - "public.users:public_users".to_string(), - "schema1.orders:schema1_orders".to_string(), - ]; - let result = super::parse_db_table_mappings(&mappings).unwrap(); - assert_eq!(result.len(), 2); - assert_eq!( - result[0], - ("public".into(), "users".into(), "public_users".into()) - ); - assert_eq!( - result[1], - ("schema1".into(), "orders".into(), "schema1_orders".into()) - ); - } - - #[test] - fn parse_db_table_mappings_missing_colon() { - let mappings = vec!["public.users".to_string()]; - let result = super::parse_db_table_mappings(&mappings); - assert!(result.is_err()); - assert!( - result - .unwrap_err() - .contains("expected schema.table:target_table") - ); - } - - #[test] - fn parse_db_table_mappings_missing_dot() { - let mappings = vec!["users:target".to_string()]; - let result = super::parse_db_table_mappings(&mappings); - assert!(result.is_err()); - assert!(result.unwrap_err().contains("expected schema.table")); - } - - #[test] - fn parse_db_table_mappings_empty() { - let mappings: Vec = vec![]; - let result = super::parse_db_table_mappings(&mappings).unwrap(); - assert!(result.is_empty()); - } - - #[test] - fn parse_enum_known_variant() { - use clickhouse_cloud_api::models::ClickPipePostObjectStorageSourceFormat; - let format: ClickPipePostObjectStorageSourceFormat = - super::parse_enum("JSONEachRow").unwrap(); - assert_eq!(format, ClickPipePostObjectStorageSourceFormat::JSONEachRow); - } - - #[test] - fn parse_enum_unknown_falls_through() { - // Unknown values map to the catch-all Unknown(String) variant — - // forwarded to the API which returns the canonical validation error. - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceType; - let kafka_type: ClickPipePostKafkaSourceType = - super::parse_enum("not-a-real-type").unwrap(); - assert_eq!( - kafka_type, - ClickPipePostKafkaSourceType::Unknown("not-a-real-type".to_string()) - ); - } - - #[test] - fn parse_enum_preserves_rename_spellings() { - // Enums use `#[serde(rename = "s3")]` etc. — wire format is authoritative. - use clickhouse_cloud_api::models::{ - ClickPipePostKafkaSourceAuthentication, ClickPipePostObjectStorageSourceType, - }; - let ty: ClickPipePostObjectStorageSourceType = super::parse_enum("s3").unwrap(); - assert_eq!(ty, ClickPipePostObjectStorageSourceType::S3); - let auth: ClickPipePostKafkaSourceAuthentication = - super::parse_enum("SCRAM-SHA-256").unwrap(); - assert_eq!(auth, ClickPipePostKafkaSourceAuthentication::SCRAM_SHA_256); - } - - #[test] - fn parse_columns_valid() { - let cols = vec!["id:Int64".to_string(), "name:String".to_string()]; - let parsed = super::parse_columns(&cols).unwrap(); - assert_eq!(parsed.len(), 2); - assert_eq!(parsed[0].name, "id"); - assert_eq!(parsed[0].r#type, "Int64"); - assert_eq!(parsed[1].name, "name"); - assert_eq!(parsed[1].r#type, "String"); - } - - #[test] - fn parse_columns_missing_colon_errors() { - let cols = vec!["id_without_type".to_string()]; - let err = super::parse_columns(&cols).unwrap_err(); - assert!(err.contains("expected name:type")); - } - - #[test] - fn build_destination_uses_defaults_for_table_definition() { - let dest = super::build_destination("mydb", "events", vec![]); - assert_eq!(dest.database, "mydb"); - assert_eq!(dest.table.as_deref(), Some("events")); - assert_eq!(dest.managed_table, Some(true)); - // Default table engine is MergeTree, not something else. - assert_eq!( - dest.table_definition - .as_ref() - .expect("non-database pipe gets a tableDefinition") - .engine - .r#type, - clickhouse_cloud_api::models::ClickPipeDestinationTableEngineType::MergeTree - ); - } - - // `build_kafka_credentials` tests — lock the wire shape for each auth mode. - // Authoritative source: `ClickPipePostKafkaSource.credentials` in - // `crates/clickhouse-cloud-api/clickhouse_cloud_openapi.json`. - - fn kafka_args() -> crate::cloud::cli::KafkaCreateArgs { - crate::cloud::cli::KafkaCreateArgs { - service_id: "svc".into(), - name: "pipe".into(), - source: crate::cloud::cli::KafkaSourceFields { - brokers: "b:9092".into(), - topics: "t".into(), - format: "JSONEachRow".into(), - kafka_type: "kafka".into(), - consumer_group: None, - auth: None, - username: None, - password: None, - iam_role: None, - access_key_id: None, - secret_key: None, - offset: "from_beginning".into(), - offset_timestamp: None, - schema_registry_url: None, - schema_registry_username: None, - schema_registry_password: None, - ca_certificate: None, - client_certificate: None, - client_key: None, - schema_registry_ca_certificate: None, - reverse_private_endpoint_ids: vec![], - }, - database: "d".into(), - table: "t".into(), - columns: vec![], - org_id: None, - } - } - - #[test] - fn kafka_credentials_plain_shape() { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - let mut args = kafka_args(); - args.source.auth = Some("PLAIN".into()); - args.source.username = Some("u".into()); - args.source.password = Some("p".into()); - let creds = super::build_kafka_credentials(&Auth::PLAIN, &args.source, None).unwrap(); - assert_eq!(creds["username"], "u"); - assert_eq!(creds["password"], "p"); - } - - #[test] - fn kafka_credentials_iam_user_shape() { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - let mut args = kafka_args(); - args.source.auth = Some("IAM_USER".into()); - args.source.access_key_id = Some("AKIA".into()); - args.source.secret_key = Some("secret".into()); - let creds = super::build_kafka_credentials(&Auth::IAM_USER, &args.source, None).unwrap(); - // MskIamUser wire shape is {accessKeyId, secretKey} — NOT snake_case. - assert_eq!(creds["accessKeyId"], "AKIA"); - assert_eq!(creds["secretKey"], "secret"); - assert!(creds.get("access_key_id").is_none()); - } - - #[test] - fn kafka_credentials_iam_role_is_null() { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - let mut args = kafka_args(); - args.source.auth = Some("IAM_ROLE".into()); - args.source.iam_role = Some("arn:aws:iam::123:role/Foo".into()); - // IAM_ROLE sends credentials=null; the role ARN flows through the - // top-level `iamRole` field on the Kafka source, not credentials. - let creds = super::build_kafka_credentials(&Auth::IAM_ROLE, &args.source, None).unwrap(); - assert!(creds.is_null()); - } - - #[test] - fn kafka_credentials_mutual_tls_shape() { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - let args = kafka_args(); - let contents = Some(("CERT_PEM".into(), "KEY_PEM".into())); - let creds = - super::build_kafka_credentials(&Auth::MUTUAL_TLS, &args.source, contents).unwrap(); - assert_eq!(creds["certificate"], "CERT_PEM"); - assert_eq!(creds["privateKey"], "KEY_PEM"); - } - - #[test] - fn kafka_credentials_iam_user_missing_args_errors() { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - let args = kafka_args(); - let err = super::build_kafka_credentials(&Auth::IAM_USER, &args.source, None).unwrap_err(); - assert!(err.contains("--access-key-id")); - } - - #[test] - fn kafka_credentials_iam_role_missing_arn_errors() { - use clickhouse_cloud_api::models::ClickPipePostKafkaSourceAuthentication as Auth; - let mut args = kafka_args(); - args.source.auth = Some("IAM_ROLE".into()); - let err = super::build_kafka_credentials(&Auth::IAM_ROLE, &args.source, None).unwrap_err(); - assert!(err.contains("--iam-role")); - } -} diff --git a/crates/clickhousectl/src/cloud/mod.rs b/crates/clickhousectl/src/cloud/mod.rs index 249b99c4..3c91588d 100644 --- a/crates/clickhousectl/src/cloud/mod.rs +++ b/crates/clickhousectl/src/cloud/mod.rs @@ -5,7 +5,6 @@ pub mod backups; pub mod cli; pub mod clickpipes; pub mod client; -pub mod commands; pub mod credentials; pub mod organizations; pub mod output; @@ -24,9 +23,7 @@ pub use client::{ }; use crate::error::{Error, Result}; -use cli::{ - ClickPipeCommands, ClickPipeCreateCommands, ClickPipeSettingsCommands, CloudArgs, CloudCommands, -}; +use cli::{CloudArgs, CloudCommands}; /// Explain when a configured environment credential cannot participate in /// authentication because a higher-precedence source won. Keep this notice on @@ -138,183 +135,7 @@ async fn dispatch( CloudCommands::Activity { command } => activity::run(client, command, json).await, CloudCommands::Backup { command } => backups::run(client, command, json).await, CloudCommands::Postgres { command } => postgres::run(client, command, json).await, - CloudCommands::ClickPipe { command } => match *command { - ClickPipeCommands::List { service_id, org_id } => { - commands::clickpipe_list(client, &service_id, org_id.as_deref(), json).await - } - ClickPipeCommands::Get { - service_id, - clickpipe_id, - org_id, - } => { - commands::clickpipe_get(client, &service_id, &clickpipe_id, org_id.as_deref(), json) - .await - } - ClickPipeCommands::Delete { - service_id, - clickpipe_id, - org_id, - } => { - commands::clickpipe_delete( - client, - &service_id, - &clickpipe_id, - org_id.as_deref(), - json, - ) - .await - } - ClickPipeCommands::Start { - service_id, - clickpipe_id, - org_id, - } => { - commands::clickpipe_state( - client, - &service_id, - &clickpipe_id, - "start", - org_id.as_deref(), - json, - ) - .await - } - ClickPipeCommands::Stop { - service_id, - clickpipe_id, - org_id, - } => { - commands::clickpipe_state( - client, - &service_id, - &clickpipe_id, - "stop", - org_id.as_deref(), - json, - ) - .await - } - ClickPipeCommands::Resync { - service_id, - clickpipe_id, - org_id, - } => { - commands::clickpipe_state( - client, - &service_id, - &clickpipe_id, - "resync", - org_id.as_deref(), - json, - ) - .await - } - ClickPipeCommands::Scale { - service_id, - clickpipe_id, - replicas, - cpu_millicores, - memory_gb, - org_id, - } => { - commands::clickpipe_scale( - client, - &service_id, - &clickpipe_id, - replicas, - cpu_millicores, - memory_gb, - org_id.as_deref(), - json, - ) - .await - } - ClickPipeCommands::Settings { command } => match command { - ClickPipeSettingsCommands::Get { - service_id, - clickpipe_id, - org_id, - } => { - commands::clickpipe_settings_get( - client, - &service_id, - &clickpipe_id, - org_id.as_deref(), - json, - ) - .await - } - ClickPipeSettingsCommands::Update { - service_id, - clickpipe_id, - streaming_max_insert_wait_ms, - object_storage_concurrency, - object_storage_polling_interval_ms, - object_storage_max_insert_bytes, - object_storage_max_file_count, - clickhouse_max_threads, - clickhouse_max_insert_threads, - object_storage_use_cluster_function, - clickhouse_parallel_view_processing, - org_id, - } => { - commands::clickpipe_settings_update( - client, - &service_id, - &clickpipe_id, - streaming_max_insert_wait_ms, - object_storage_concurrency, - object_storage_polling_interval_ms, - object_storage_max_insert_bytes, - object_storage_max_file_count, - clickhouse_max_threads, - clickhouse_max_insert_threads, - object_storage_use_cluster_function, - clickhouse_parallel_view_processing, - org_id.as_deref(), - json, - ) - .await - } - }, - ClickPipeCommands::SchemaDiscover { - service_id, - command, - org_id, - } => { - commands::clickpipe_schema_discover( - client, - &service_id, - &command, - org_id.as_deref(), - json, - ) - .await - } - ClickPipeCommands::Create { command } => match command { - ClickPipeCreateCommands::ObjectStorage(args) => { - commands::clickpipe_create_s3(client, &args, json).await - } - ClickPipeCreateCommands::Kafka(args) => { - commands::clickpipe_create_kafka(client, &args, json).await - } - ClickPipeCreateCommands::Kinesis(args) => { - commands::clickpipe_create_kinesis(client, &args, json).await - } - ClickPipeCreateCommands::Postgres(args) => { - commands::clickpipe_create_postgres(client, &args, json).await - } - ClickPipeCreateCommands::MySQL(args) => { - commands::clickpipe_create_mysql(client, &args, json).await - } - ClickPipeCreateCommands::MongoDB(args) => { - commands::clickpipe_create_mongodb(client, &args, json).await - } - ClickPipeCreateCommands::BigQuery(args) => { - commands::clickpipe_create_bigquery(client, &args, json).await - } - }, - }, + CloudCommands::ClickPipe { command } => clickpipes::run(client, *command, json).await, } }