diff --git a/crates/flowctl/README.md b/crates/flowctl/README.md index 2ecb50719a7..cae7f849a91 100644 --- a/crates/flowctl/README.md +++ b/crates/flowctl/README.md @@ -55,3 +55,36 @@ flowctl draft create flowctl draft author --source ~/estuary/flow/examples/citi-bike/flow.yaml flowctl draft publish ``` + +### Working with the GraphQL API directly + +`flowctl raw graphql` explores the control-plane GraphQL API and runs +operations against it. The schema is read by introspection, so it always +describes the API the active profile points at. + +```console +# What can I call? +flowctl raw graphql operations +flowctl raw graphql operations mutation --search invite + +# What shape is this type? +flowctl raw graphql describe InviteLink +flowctl raw graphql types --kind input-object --search alert + +# The whole schema, as SDL. +flowctl raw graphql schema +``` + +Run a query or mutation with `exec`. The document comes from an argument, +from `--file`, or from stdin, and variables from `--variables` (a JSON +object) or repeated `--var name=value` pairs: + +```console +flowctl raw graphql exec 'query { alertTypes { alertType } }' + +flowctl raw graphql exec --var cap=read \ + 'query Q($cap: Capability!) { prefixes(by: {minCapability: $cap}) { edges { node { prefix } } } }' +``` + +The full response is printed as JSON, including its `errors`, and the +command exits non-zero when the API reported any. diff --git a/crates/flowctl/src/graphql.rs b/crates/flowctl/src/graphql.rs index be39bd21d1b..c25f5aaf968 100644 --- a/crates/flowctl/src/graphql.rs +++ b/crates/flowctl/src/graphql.rs @@ -2,6 +2,9 @@ //! //! This module contains some common types for use with graphql queries. //! +//! To explore the schema, or to run an operation without writing a typed query +//! first, see the `flowctl raw graphql` commands in `crate::raw::graphql`. +//! //! We use the `graphql_client` crate for all graphql requests: //! https://github.com/graphql-rust/graphql-client //! diff --git a/crates/flowctl/src/raw/graphql/introspection.rs b/crates/flowctl/src/raw/graphql/introspection.rs new file mode 100644 index 00000000000..51bea893251 --- /dev/null +++ b/crates/flowctl/src/raw/graphql/introspection.rs @@ -0,0 +1,360 @@ +//! POD types mirroring a GraphQL introspection response, and the query which +//! produces one. +//! +//! Introspection is the GraphQL-native way to ask a server what it serves: the +//! `__schema` meta-field returns every type, field, argument, and directive of +//! the running schema. `flowctl` asks the control-plane API directly rather than +//! reading the checked-in `control-plane-api.graphql`, so that these commands +//! describe the API the user is actually talking to. +//! +//! The shapes here follow the GraphQL specification's introspection schema, so +//! they deserialize the response of [`QUERY`] and of any other spec-conformant +//! server. + +/// Full-schema introspection query. +/// +/// `includeDeprecated: true` is passed everywhere it's accepted so that +/// deprecated fields, enum values, input fields, and arguments are described +/// instead of silently omitted. `TypeRef` is unrolled to eight levels of +/// wrapping types, which is the depth conventionally used to cover any +/// realistic nesting of `[T!]!`. +pub const QUERY: &str = r#" +query FlowctlIntrospection { + __schema { + queryType { name } + mutationType { name } + subscriptionType { name } + types { ...FullType } + directives { + name + description + locations + args(includeDeprecated: true) { ...InputValue } + } + } +} + +fragment FullType on __Type { + kind + name + description + fields(includeDeprecated: true) { + name + description + args(includeDeprecated: true) { ...InputValue } + type { ...TypeRef } + isDeprecated + deprecationReason + } + inputFields(includeDeprecated: true) { ...InputValue } + isOneOf + interfaces { ...TypeRef } + enumValues(includeDeprecated: true) { + name + description + isDeprecated + deprecationReason + } + possibleTypes { ...TypeRef } +} + +fragment InputValue on __InputValue { + name + description + type { ...TypeRef } + defaultValue + isDeprecated + deprecationReason +} + +fragment TypeRef on __Type { + kind + name + ofType { + kind + name + ofType { + kind + name + ofType { + kind + name + ofType { + kind + name + ofType { + kind + name + ofType { + kind + name + ofType { kind name } + } + } + } + } + } + } +} +"#; + +/// Data of an introspection response. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +pub struct Data { + #[serde(rename = "__schema")] + pub schema: Schema, +} + +/// A GraphQL schema: its root operation types, every named type, and every +/// directive. +/// +/// Every optional field here defaults, because a response only carries the +/// fields its query selected: a narrower introspection query than [`QUERY`] +/// still deserializes into these types. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Schema { + #[serde(default)] + pub query_type: Option, + #[serde(default)] + pub mutation_type: Option, + #[serde(default)] + pub subscription_type: Option, + pub types: Vec, + #[serde(default)] + pub directives: Vec, +} + +/// Reference to a root operation type, which is always a named object type. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +pub struct NamedType { + pub name: String, +} + +/// A named type of the schema. Which of the optional collections are populated +/// depends on `kind`: `fields` for objects and interfaces, `input_fields` for +/// input objects, `enum_values` for enums, and so on. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Type { + pub kind: Kind, + #[serde(default)] + pub name: Option, + #[serde(default)] + pub description: Option, + #[serde(default)] + pub fields: Option>, + #[serde(default)] + pub input_fields: Option>, + /// Set on an input object which requires that exactly one of its fields be + /// given, and which SDL marks with `@oneOf`. + #[serde(default)] + pub is_one_of: Option, + #[serde(default)] + pub interfaces: Option>, + #[serde(default)] + pub enum_values: Option>, + #[serde(default)] + pub possible_types: Option>, +} + +/// Kind of a type. `LIST` and `NON_NULL` are wrapping kinds which only ever +/// appear within a [`TypeRef`]. +#[derive( + Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize, serde::Serialize, clap::ValueEnum, +)] +#[serde(rename_all = "SCREAMING_SNAKE_CASE")] +pub enum Kind { + Scalar, + Object, + Interface, + Union, + Enum, + InputObject, + List, + NonNull, +} + +/// A field of an object or interface type. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct Field { + pub name: String, + #[serde(default)] + pub description: Option, + #[serde(default)] + pub args: Vec, + #[serde(rename = "type")] + pub of_type: TypeRef, + #[serde(default)] + pub is_deprecated: bool, + #[serde(default)] + pub deprecation_reason: Option, +} + +/// An argument of a field or directive, or a field of an input object. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct InputValue { + pub name: String, + #[serde(default)] + pub description: Option, + #[serde(rename = "type")] + pub of_type: TypeRef, + #[serde(default)] + pub default_value: Option, + #[serde(default)] + pub is_deprecated: bool, + #[serde(default)] + pub deprecation_reason: Option, +} + +/// A value of an enum type. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct EnumValue { + pub name: String, + #[serde(default)] + pub description: Option, + #[serde(default)] + pub is_deprecated: bool, + #[serde(default)] + pub deprecation_reason: Option, +} + +/// A directive the schema understands, such as `@deprecated`. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +pub struct Directive { + pub name: String, + #[serde(default)] + pub description: Option, + #[serde(default)] + pub locations: Vec, + #[serde(default)] + pub args: Vec, +} + +/// A use of a type, which wraps a named type in any number of `LIST` and +/// `NON_NULL` kinds. `[Foo!]!` arrives as `NON_NULL(LIST(NON_NULL(Foo)))`. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TypeRef { + pub kind: Kind, + #[serde(default)] + pub name: Option, + #[serde(default)] + pub of_type: Option>, +} + +impl std::fmt::Display for TypeRef { + /// Renders the reference in GraphQL type syntax, as `[Foo!]!`. + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + // A wrapping kind without an inner type is not spec-conformant. Render a + // placeholder rather than panicking, so that introspecting a + // non-conformant server still produces usable output. + match (self.kind, &self.of_type) { + (Kind::NonNull, Some(inner)) => write!(f, "{inner}!"), + (Kind::List, Some(inner)) => write!(f, "[{inner}]"), + _ => f.write_str(self.name.as_deref().unwrap_or("")), + } + } +} + +impl Kind { + /// The SDL keyword which introduces a type of this kind. + pub fn keyword(&self) -> &'static str { + match self { + Kind::Scalar => "scalar", + Kind::Object => "type", + Kind::Interface => "interface", + Kind::Union => "union", + Kind::Enum => "enum", + Kind::InputObject => "input", + // Wrapping kinds are never introduced as a named definition. + Kind::List => "list", + Kind::NonNull => "non-null", + } + } +} + +impl std::fmt::Display for Kind { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Kind::Scalar => "SCALAR", + Kind::Object => "OBJECT", + Kind::Interface => "INTERFACE", + Kind::Union => "UNION", + Kind::Enum => "ENUM", + Kind::InputObject => "INPUT_OBJECT", + Kind::List => "LIST", + Kind::NonNull => "NON_NULL", + }) + } +} + +/// Scalars every GraphQL server defines. They carry no information for a user +/// exploring this API, so listings hide them by default. +pub const BUILT_IN_SCALARS: [&str; 5] = ["Boolean", "Float", "ID", "Int", "String"]; + +impl Schema { + /// Looks up a named type, case-insensitively so that `flowctl raw graphql + /// describe livespec` finds `LiveSpec`. + pub fn find_type(&self, name: &str) -> Option<&Type> { + self.types + .iter() + .find(|ty| ty.name.as_deref() == Some(name)) + .or_else(|| { + self.types.iter().find(|ty| { + ty.name + .as_deref() + .is_some_and(|n| n.eq_ignore_ascii_case(name)) + }) + }) + } + + /// Types of the schema, sorted by name, less the introspection meta types + /// (`__Schema`, `__Type`, ...) and built-in scalars unless `include_builtins`. + pub fn named_types(&self, include_builtins: bool) -> Vec<&Type> { + let mut types: Vec<&Type> = self + .types + .iter() + .filter(|ty| { + let Some(name) = ty.name.as_deref() else { + return false; + }; + include_builtins || !(name.starts_with("__") || BUILT_IN_SCALARS.contains(&name)) + }) + .collect(); + + types.sort_by_key(|ty| ty.name.as_deref().unwrap_or_default()); + types + } + + /// The object type backing an operation root, if the schema defines that root. + pub fn root_type(&self, operation: Operation) -> Option<&Type> { + let root = match operation { + Operation::Query => self.query_type.as_ref(), + Operation::Mutation => self.mutation_type.as_ref(), + }; + self.find_type(&root?.name) + } +} + +/// A root operation of the schema which `flowctl` can invoke. Subscriptions are +/// omitted because the control-plane API serves none, and because they need a +/// streaming transport rather than the unary POST used here. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Serialize, clap::ValueEnum)] +#[serde(rename_all = "camelCase")] +pub enum Operation { + Query, + Mutation, +} + +impl std::fmt::Display for Operation { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str(match self { + Operation::Query => "query", + Operation::Mutation => "mutation", + }) + } +} diff --git a/crates/flowctl/src/raw/graphql/mod.rs b/crates/flowctl/src/raw/graphql/mod.rs new file mode 100644 index 00000000000..066d13dfb80 --- /dev/null +++ b/crates/flowctl/src/raw/graphql/mod.rs @@ -0,0 +1,502 @@ +//! Ad-hoc access to the control-plane GraphQL API. +//! +//! These commands are the GraphQL counterpart of `flowctl raw get` and `flowctl +//! raw rpc`: they let you call the API directly, without a purpose-built +//! `flowctl` command, and they let you ask the API to describe itself. +//! +//! `exec` posts a document you supply. The remaining commands read the schema by +//! introspection and render it: `schema` for the whole SDL, `types` and +//! `operations` for listings, and `describe` for one type. + +use anyhow::Context; +use itertools::Itertools; +use std::io::Write; + +mod introspection; +mod render; + +#[derive(Debug, clap::Args)] +#[clap(rename_all = "kebab-case")] +pub struct Graphql { + #[clap(subcommand)] + cmd: Command, +} + +#[derive(Debug, clap::Subcommand)] +#[clap(rename_all = "kebab-case")] +pub enum Command { + /// Execute a GraphQL query or mutation. + /// + /// The document is taken from the positional argument, from --file, or from + /// stdin. Variables are given as a JSON object with --variables, as + /// individual --var name=value pairs, or both. + /// + /// The complete response is printed, including its `errors` if the API + /// returned any, and the command exits non-zero when it did. + /// + /// For example: + /// + /// flowctl raw graphql exec 'query { alertTypes { alertType } }' + /// + /// flowctl raw graphql exec --var prefix=acmeCo/ \ + /// 'query Q($prefix: Prefix!) { prefixes(by: {prefix: $prefix}) { edges { node { prefix } } } }' + Exec(Exec), + /// Print the schema which the API serves. + /// + /// The schema is read by introspection, so it describes the API this profile + /// is pointed at. SDL output follows the conventions of the schema checked in + /// at crates/flow-client/control-plane-api.graphql, so the two can be diffed + /// to find where a client's generated types have fallen behind the API. + Schema(Schema), + /// List the types of the schema. + Types(Types), + /// Print the definition of one type, as SDL. + Describe(Describe), + /// List the queries and mutations which the API serves. + Operations(Operations), +} + +#[derive(Debug, clap::Args)] +#[clap(rename_all = "kebab-case")] +pub struct Exec { + /// GraphQL document to execute. + /// + /// Read from stdin when omitted, or when it is `-`. + document: Option, + /// Path of a file holding the GraphQL document to execute. + #[clap(long, conflicts_with = "document")] + file: Option, + /// Variables of the operation, as a JSON object. + #[clap(long)] + variables: Option, + /// A single variable, as `name=value`, which may be repeated. + /// + /// The value is used as JSON if it parses as JSON, and as a string + /// otherwise: `--var first=10` passes a number, and `--var prefix=acmeCo/` + /// passes a string. Pairs given here override --variables. + #[clap(long = "var", value_parser = super::parse_key_val::, number_of_values = 1)] + var: Vec<(String, String)>, + /// Name of the operation to execute. + /// + /// Required when the document defines more than one operation. + #[clap(long)] + operation_name: Option, +} + +#[derive(Debug, clap::Args)] +#[clap(rename_all = "kebab-case")] +pub struct Schema { + /// Form in which to print the schema. + #[clap(long, value_enum, default_value = "sdl")] + format: SchemaFormat, + /// Also include built-in scalars and the `__`-prefixed introspection types. + #[clap(long)] + all: bool, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, clap::ValueEnum)] +#[clap(rename_all = "kebab-case")] +pub enum SchemaFormat { + /// GraphQL schema definition language. + Sdl, + /// The introspection response itself, which is what other GraphQL tooling + /// consumes. + Introspection, +} + +#[derive(Debug, clap::Args)] +#[clap(rename_all = "kebab-case")] +pub struct Types { + /// Only list types of this kind. + #[clap(long, value_enum)] + kind: Option, + /// Only list types whose name contains this substring, case-insensitively. + #[clap(long)] + search: Option, + /// Also list built-in scalars and the `__`-prefixed introspection types. + #[clap(long)] + all: bool, +} + +#[derive(Debug, clap::Args)] +#[clap(rename_all = "kebab-case")] +pub struct Describe { + /// Name of the type to describe, such as `QueryRoot` or `LiveSpec`. + /// + /// Matched case-insensitively when there's no exact match. + name: String, +} + +#[derive(Debug, clap::Args)] +#[clap(rename_all = "kebab-case")] +pub struct Operations { + /// Limit the listing to one root: `query` or `mutation`. + #[clap(value_enum)] + operation: Option, + /// Only list operations whose name contains this substring, case-insensitively. + #[clap(long)] + search: Option, +} + +pub async fn do_graphql(ctx: &mut crate::CliContext, args: &Graphql) -> anyhow::Result<()> { + match &args.cmd { + Command::Exec(exec) => do_exec(ctx, exec).await, + Command::Schema(schema) => do_schema(ctx, schema).await, + Command::Types(types) => do_types(ctx, types).await, + Command::Describe(describe) => do_describe(ctx, describe).await, + Command::Operations(operations) => do_operations(ctx, operations).await, + } +} + +/// A GraphQL request, as posted to the API. +#[derive(Debug, serde::Serialize)] +struct Request<'a> { + query: &'a str, + #[serde(skip_serializing_if = "Option::is_none")] + variables: Option, + #[serde(rename = "operationName", skip_serializing_if = "Option::is_none")] + operation_name: Option<&'a str>, +} + +async fn do_exec(ctx: &mut crate::CliContext, exec: &Exec) -> anyhow::Result<()> { + let document = read_document(exec)?; + if document.trim().is_empty() { + anyhow::bail!("no GraphQL document was provided"); + } + let variables = build_variables(exec.variables.as_deref(), &exec.var)?; + + // The whole response envelope is wanted here, errors included, so it's + // deserialized as an opaque JSON value rather than through `post_graphql`. + let response: serde_json::Value = crate::graphql::agent_unary( + &ctx.rest, + ctx.access_token().as_deref(), + crate::graphql::GRAPHQL_PATH, + &Request { + query: &document, + variables, + operation_name: exec.operation_name.as_deref(), + }, + ) + .await + .context("executing the GraphQL request")?; + + print_json_or_yaml(ctx, &response)?; + + // Signal failure through the exit code, since a GraphQL error arrives with an + // HTTP 200 and would otherwise look like success to a calling script. + let errors = response + .get("errors") + .and_then(serde_json::Value::as_array) + .map(Vec::as_slice) + .unwrap_or_default(); + + if !errors.is_empty() { + anyhow::bail!( + "the API returned {} GraphQL error(s): {}", + errors.len(), + errors + .iter() + .map(|error| error + .get("message") + .and_then(serde_json::Value::as_str) + .unwrap_or("(no message)")) + .format("; ") + ); + } + Ok(()) +} + +async fn do_schema(ctx: &mut crate::CliContext, args: &Schema) -> anyhow::Result<()> { + if let SchemaFormat::Introspection = args.format { + let response = introspect_raw(ctx).await?; + return print_json_or_yaml(ctx, &response); + } + let schema = introspect(ctx).await?; + + let mut stdout = std::io::stdout().lock(); + stdout.write_all(render::schema(&schema, args.all).as_bytes())?; + Ok(()) +} + +async fn do_types(ctx: &mut crate::CliContext, args: &Types) -> anyhow::Result<()> { + let schema = introspect(ctx).await?; + + let rows: Vec = schema + .named_types(args.all) + .into_iter() + .filter(|ty| args.kind.is_none_or(|kind| kind == ty.kind)) + .filter(|ty| matches_search(ty.name.as_deref(), args.search.as_deref())) + .map(render::TypeSummary::new) + .collect(); + + if rows.is_empty() { + anyhow::bail!("no types of the schema match this filter"); + } + ctx.write_all(rows, ()) +} + +async fn do_describe(ctx: &mut crate::CliContext, args: &Describe) -> anyhow::Result<()> { + let schema = introspect(ctx).await?; + + let Some(ty) = schema.find_type(&args.name) else { + // Point at the near misses, since type names are long and camel-cased. + let similar = similar_names(&schema, &args.name); + + if similar.is_empty() { + anyhow::bail!( + "the schema has no type named {:?}. Run `flowctl raw graphql types` to list them", + args.name + ); + } + anyhow::bail!( + "the schema has no type named {:?}. Similar names: {}", + args.name, + similar.iter().format(", ") + ); + }; + + let mut stdout = std::io::stdout().lock(); + stdout.write_all(render::type_def(ty).as_bytes())?; + Ok(()) +} + +async fn do_operations(ctx: &mut crate::CliContext, args: &Operations) -> anyhow::Result<()> { + let schema = introspect(ctx).await?; + + let operations = match args.operation { + Some(operation) => vec![operation], + None => vec![ + introspection::Operation::Query, + introspection::Operation::Mutation, + ], + }; + + let mut rows: Vec = Vec::new(); + for operation in operations { + let Some(root) = schema.root_type(operation) else { + tracing::debug!(%operation, "the schema defines no root type for this operation"); + continue; + }; + rows.extend( + root.fields + .iter() + .flatten() + .filter(|field| matches_search(Some(&field.name), args.search.as_deref())) + .map(|field| render::OperationSummary::new(operation, field)), + ); + } + + if rows.is_empty() { + anyhow::bail!("no operations of the schema match this filter"); + } + ctx.write_all(rows, ()) +} + +/// Reads the schema of the API by introspection. +async fn introspect(ctx: &crate::CliContext) -> anyhow::Result { + let response: graphql_client::Response = crate::graphql::agent_unary( + &ctx.rest, + ctx.access_token().as_deref(), + crate::graphql::GRAPHQL_PATH, + &Request { + query: introspection::QUERY, + variables: None, + operation_name: None, + }, + ) + .await + .context("introspecting the GraphQL API")?; + + if let Some(errors) = response.errors.filter(|errors| !errors.is_empty()) { + anyhow::bail!("introspection failed: [{}]", errors.iter().format(", ")); + } + Ok(response + .data + .context("the introspection response has no data")? + .schema) +} + +/// Reads the schema of the API by introspection, without interpreting the +/// response. Backs `schema --format json`, which is the form other GraphQL +/// tooling consumes. +async fn introspect_raw(ctx: &crate::CliContext) -> anyhow::Result { + crate::graphql::agent_unary( + &ctx.rest, + ctx.access_token().as_deref(), + crate::graphql::GRAPHQL_PATH, + &Request { + query: introspection::QUERY, + variables: None, + operation_name: None, + }, + ) + .await + .context("introspecting the GraphQL API") +} + +fn read_document(exec: &Exec) -> anyhow::Result { + if let Some(path) = &exec.file { + return std::fs::read_to_string(path) + .with_context(|| format!("reading {}", path.display())); + } + match exec.document.as_deref() { + Some("-") | None => { + let mut document = String::new(); + std::io::Read::read_to_string(&mut std::io::stdin().lock(), &mut document) + .context("reading the GraphQL document from stdin")?; + Ok(document) + } + Some(document) => Ok(document.to_string()), + } +} + +/// Merges `--variables` with repeated `--var name=value` pairs, which take +/// precedence. A pair's value is used as JSON when it parses as JSON, and as a +/// string otherwise, so that both `--var first=10` and `--var prefix=acmeCo/` +/// pass the value the caller intends. +fn build_variables( + variables: Option<&str>, + var: &[(String, String)], +) -> anyhow::Result> { + let mut merged = match variables { + Some(variables) => match serde_json::from_str(variables).context("parsing --variables")? { + serde_json::Value::Object(object) => object, + other => anyhow::bail!("--variables must be a JSON object, but is {other}"), + }, + None => serde_json::Map::new(), + }; + + for (name, value) in var { + let value = serde_json::from_str(value) + .unwrap_or_else(|_| serde_json::Value::String(value.clone())); + merged.insert(name.clone(), value); + } + + if merged.is_empty() { + return Ok(None); + } + Ok(Some(serde_json::Value::Object(merged))) +} + +/// Names of `schema`'s types which are similar to `name`, for the error message +/// of a failed lookup. +/// +/// Containment is tested in both directions, so that an under-specified guess +/// (`invite`) and an over-specified one (`InviteLinkResult`) both surface +/// `InviteLink`. The reverse direction is limited to candidates of at least four +/// characters, because a short name such as `Id` is a substring of too many +/// guesses to be a useful suggestion. +fn similar_names<'a>(schema: &'a introspection::Schema, name: &str) -> Vec<&'a str> { + const MIN_CONTAINED: usize = 4; + let name = name.to_lowercase(); + + schema + .named_types(false) + .into_iter() + .filter_map(|ty| ty.name.as_deref()) + .filter(|candidate| { + let candidate = candidate.to_lowercase(); + candidate.contains(&name) + || (candidate.len() >= MIN_CONTAINED && name.contains(&candidate)) + }) + .take(10) + .collect() +} + +fn matches_search(name: Option<&str>, search: Option<&str>) -> bool { + match (name, search) { + (_, None) => true, + (Some(name), Some(search)) => name.to_lowercase().contains(&search.to_lowercase()), + (None, Some(_)) => false, + } +} + +/// Prints a GraphQL response or introspection payload. +/// +/// JSON is the default because it's the encoding GraphQL itself uses, and it's +/// what a pipe into `jq` expects. Only an explicit `--output yaml` changes it: +/// there are no fixed columns to build a table from, and the usual "YAML when +/// stdout isn't a terminal" default would otherwise break those pipes. +fn print_json_or_yaml( + ctx: &mut crate::CliContext, + value: &serde_json::Value, +) -> anyhow::Result<()> { + let mut stdout = std::io::stdout().lock(); + + match ctx.output.output { + Some(crate::output::OutputType::Yaml) => serde_yaml::to_writer(&mut stdout, value)?, + _ => serde_json::to_writer_pretty(&mut stdout, value)?, + } + stdout.write_all(b"\n")?; + Ok(()) +} + +#[cfg(test)] +mod test { + use super::*; + + fn var(name: &str, value: &str) -> (String, String) { + (name.to_string(), value.to_string()) + } + + #[test] + fn test_build_variables() { + // No inputs at all omits `variables` from the request. + assert_eq!(build_variables(None, &[]).unwrap(), None); + + // A `--var` value is JSON when it parses as JSON, and a string otherwise. + let vars = [ + var("first", "10"), + var("prefix", "acmeCo/"), + var("closed", "true"), + var("filter", r#"{"catalogPrefix": {"startsWith": "acmeCo/"}}"#), + var("quoted", r#""10""#), + ]; + insta::assert_json_snapshot!(build_variables(None, &vars).unwrap()); + + // `--var` pairs are merged over `--variables`. + insta::assert_json_snapshot!( + build_variables( + Some(r#"{"first": 5, "prefix": "wileyCo/"}"#), + &[var("first", "10")] + ) + .unwrap() + ); + + // `--variables` must hold an object, and must be valid JSON. + insta::assert_snapshot!( + build_variables(Some("[1, 2]"), &[]).unwrap_err(), + @"--variables must be a JSON object, but is [1,2]" + ); + insta::assert_snapshot!( + build_variables(Some("{"), &[]).unwrap_err(), + @"parsing --variables" + ); + } + + #[test] + fn test_similar_names() { + let schema = render::test::schema_fixture(); + + // An over-specified guess finds the name it extends, and an + // under-specified one finds every name containing it. + assert_eq!(similar_names(&schema, "LiveSpecFoo"), vec!["LiveSpec"]); + assert_eq!( + similar_names(&schema, "input"), + vec!["AwsInput", "PrivateLinkConfigInput"] + ); + + // `Id` is too short to offer merely because the guess contains it. + assert!(similar_names(&schema, "Idle").is_empty()); + assert!(similar_names(&schema, "Zebra").is_empty()); + } + + #[test] + fn test_matches_search() { + assert!(matches_search(Some("LiveSpec"), None)); + assert!(matches_search(Some("LiveSpec"), Some("livespec"))); + assert!(matches_search(Some("LiveSpecConnection"), Some("Spec"))); + assert!(!matches_search(Some("LiveSpec"), Some("alert"))); + assert!(!matches_search(None, Some("alert"))); + } +} diff --git a/crates/flowctl/src/raw/graphql/render.rs b/crates/flowctl/src/raw/graphql/render.rs new file mode 100644 index 00000000000..031249ccc98 --- /dev/null +++ b/crates/flowctl/src/raw/graphql/render.rs @@ -0,0 +1,525 @@ +//! Pure rendering of an introspected schema: SDL text, and the row-shaped +//! summaries used by the `types` and `operations` listings. +//! +//! SDL output follows the conventions of the schema which the control-plane API +//! emits into `crates/flow-client/control-plane-api.graphql` — tab indentation, +//! block descriptions, types sorted by name, then directives, then the `schema` +//! block. Output of `flowctl raw graphql schema` is therefore close enough to +//! that file to diff against it, which is how a client learns whether its +//! generated types still match the API it's calling. The one deliberate +//! departure is that a documented argument list puts every argument on its own +//! line; `async_graphql`'s own SDL export only breaks the line before arguments +//! which carry a description, and runs the rest together. + +use super::introspection::{ + Directive, EnumValue, Field, InputValue, Kind, Operation, Schema, Type, TypeRef, +}; + +/// Renders the whole schema as SDL. `include_builtins` also emits the +/// introspection meta types and built-in scalars. +pub fn schema(schema: &Schema, include_builtins: bool) -> String { + let mut out = String::new(); + + for ty in schema.named_types(include_builtins) { + out.push_str(&type_def(ty)); + out.push('\n'); + } + for directive in &schema.directives { + out.push_str(&directive_def(directive)); + } + + // The root operation types are named `QueryRoot` / `MutationRoot` rather than + // the defaults, so the `schema` block is required to bind them. + out.push_str("schema {\n"); + if let Some(root) = &schema.query_type { + out.push_str(&format!("\tquery: {}\n", root.name)); + } + if let Some(root) = &schema.mutation_type { + out.push_str(&format!("\tmutation: {}\n", root.name)); + } + if let Some(root) = &schema.subscription_type { + out.push_str(&format!("\tsubscription: {}\n", root.name)); + } + out.push_str("}\n"); + + out +} + +/// Renders one named type as an SDL definition. +pub fn type_def(ty: &Type) -> String { + let mut out = String::new(); + let name = ty.name.as_deref().unwrap_or(""); + + description(&mut out, ty.description.as_deref(), 0); + + match ty.kind { + Kind::Scalar => { + out.push_str(&format!("scalar {name}\n")); + } + Kind::Union => { + let members = ty + .possible_types + .iter() + .flatten() + .map(TypeRef::to_string) + .collect::>() + .join(" | "); + out.push_str(&format!("union {name} = {members}\n")); + } + Kind::Enum => { + out.push_str(&format!("enum {name} {{\n")); + for value in ty.enum_values.iter().flatten() { + enum_value(&mut out, value); + } + out.push_str("}\n"); + } + Kind::InputObject => { + let one_of = if ty.is_one_of == Some(true) { + " @oneOf" + } else { + "" + }; + out.push_str(&format!("input {name}{one_of} {{\n")); + for input in ty.input_fields.iter().flatten() { + description(&mut out, input.description.as_deref(), 1); + out.push_str(&format!("\t{}\n", input_value(input))); + } + out.push_str("}\n"); + } + Kind::Object | Kind::Interface => { + let implements: Vec = ty + .interfaces + .iter() + .flatten() + .map(TypeRef::to_string) + .collect(); + + out.push_str(ty.kind.keyword()); + out.push_str(&format!(" {name}")); + if !implements.is_empty() { + out.push_str(&format!(" implements {}", implements.join(" & "))); + } + out.push_str(" {\n"); + + for f in ty.fields.iter().flatten() { + field(&mut out, f); + } + out.push_str("}\n"); + } + // Wrapping kinds are never top-level definitions, but a non-conformant + // server could report one. Emit something readable instead of nothing. + Kind::List | Kind::NonNull => { + out.push_str(&format!("# {} {name}\n", ty.kind)); + } + } + + out +} + +/// Renders a directive definition, as `directive @skip(if: Boolean!) on FIELD`. +fn directive_def(directive: &Directive) -> String { + let mut out = String::new(); + description(&mut out, directive.description.as_deref(), 0); + + let args = if directive.args.is_empty() { + String::new() + } else { + format!( + "({})", + directive + .args + .iter() + .map(input_value) + .collect::>() + .join(", ") + ) + }; + out.push_str(&format!( + "directive @{}{args} on {}\n", + directive.name, + directive.locations.join(" | ") + )); + out +} + +/// Renders a field of an object or interface, with its arguments. Arguments go +/// on their own lines when any of them is documented, so that the descriptions +/// stay readable. +fn field(out: &mut String, field: &Field) { + description(out, field.description.as_deref(), 1); + + let multiline = field.args.iter().any(|arg| arg.description.is_some()); + + if field.args.is_empty() { + out.push_str(&format!("\t{}", field.name)); + } else if multiline { + out.push_str(&format!("\t{}(\n", field.name)); + for arg in &field.args { + description(out, arg.description.as_deref(), 2); + out.push_str(&format!("\t\t{}\n", input_value(arg))); + } + out.push_str("\t)"); + } else { + let args = field + .args + .iter() + .map(input_value) + .collect::>() + .join(", "); + out.push_str(&format!("\t{}({args})", field.name)); + } + + out.push_str(&format!(": {}", field.of_type)); + if field.is_deprecated { + out.push_str(&deprecated(field.deprecation_reason.as_deref())); + } + out.push('\n'); +} + +/// Renders an argument or input-object field, as `first: Int = 10`. +fn input_value(input: &InputValue) -> String { + let mut out = format!("{}: {}", input.name, input.of_type); + + // `default_value` arrives as a GraphQL literal, already quoted if a string. + if let Some(default) = &input.default_value { + out.push_str(&format!(" = {default}")); + } + if input.is_deprecated { + out.push_str(&deprecated(input.deprecation_reason.as_deref())); + } + out +} + +fn enum_value(out: &mut String, value: &EnumValue) { + description(out, value.description.as_deref(), 1); + out.push_str(&format!("\t{}", value.name)); + if value.is_deprecated { + out.push_str(&deprecated(value.deprecation_reason.as_deref())); + } + out.push('\n'); +} + +fn deprecated(reason: Option<&str>) -> String { + match reason { + // GraphQL and JSON agree on string escaping for everything a schema can + // hold, so serde_json produces a valid GraphQL string literal. + Some(reason) => format!( + " @deprecated(reason: {})", + serde_json::to_string(reason).expect("a str always serializes") + ), + None => " @deprecated".to_string(), + } +} + +/// Writes a block description at `indent` tabs, or nothing when undocumented. +fn description(out: &mut String, description: Option<&str>, indent: usize) { + let Some(description) = description else { + return; + }; + let tabs = "\t".repeat(indent); + + out.push_str(&format!("{tabs}\"\"\"\n")); + for line in description.lines() { + out.push_str(&format!("{tabs}{line}\n")); + } + out.push_str(&format!("{tabs}\"\"\"\n")); +} + +/// One row of `flowctl raw graphql types`. +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct TypeSummary { + pub kind: Kind, + pub name: String, + pub description: Option, + /// Count of fields, input fields, enum values, or union members. + pub members: usize, +} + +impl TypeSummary { + pub fn new(ty: &Type) -> Self { + Self { + kind: ty.kind, + name: ty.name.clone().unwrap_or_default(), + description: ty.description.clone(), + members: ty.fields.iter().flatten().count() + + ty.input_fields.iter().flatten().count() + + ty.enum_values.iter().flatten().count() + + ty.possible_types.iter().flatten().count(), + } + } +} + +impl crate::output::CliOutput for TypeSummary { + type TableAlt = (); + type CellValue = String; + + fn table_headers(_alt: Self::TableAlt) -> Vec<&'static str> { + vec!["Kind", "Name", "Members", "Description"] + } + + fn into_table_row(self, _alt: Self::TableAlt) -> Vec { + vec![ + self.kind.to_string(), + self.name, + self.members.to_string(), + summarize(self.description.as_deref()), + ] + } +} + +/// One row of `flowctl raw graphql operations`: a field of the query or mutation +/// root, which is a thing the caller can invoke. +#[derive(Debug, serde::Serialize)] +#[serde(rename_all = "camelCase")] +pub struct OperationSummary { + pub operation: Operation, + pub name: String, + pub arguments: String, + pub returns: String, + pub description: Option, + pub deprecated: Option, +} + +impl OperationSummary { + pub fn new(operation: Operation, field: &Field) -> Self { + Self { + operation, + name: field.name.clone(), + arguments: field + .args + .iter() + .map(input_value) + .collect::>() + .join(", "), + returns: field.of_type.to_string(), + description: field.description.clone(), + deprecated: field.is_deprecated.then(|| { + field + .deprecation_reason + .clone() + .unwrap_or_else(|| "yes".to_string()) + }), + } + } +} + +impl crate::output::CliOutput for OperationSummary { + type TableAlt = (); + type CellValue = String; + + fn table_headers(_alt: Self::TableAlt) -> Vec<&'static str> { + vec!["Operation", "Name", "Arguments", "Returns", "Description"] + } + + fn into_table_row(self, _alt: Self::TableAlt) -> Vec { + let description = match &self.deprecated { + Some(reason) => format!("DEPRECATED: {reason}"), + None => summarize(self.description.as_deref()), + }; + vec![ + self.operation.to_string(), + self.name, + self.arguments, + self.returns, + description, + ] + } +} + +/// First line of a description, shortened for a table cell. +fn summarize(description: Option<&str>) -> String { + const MAX: usize = 96; + + let Some(first) = description.and_then(|d| d.lines().find(|l| !l.trim().is_empty())) else { + return String::new(); + }; + let first = first.trim(); + + match first.char_indices().nth(MAX) { + Some((offset, _)) => format!("{}…", &first[..offset]), + None => first.to_string(), + } +} + +#[cfg(test)] +pub(crate) mod test { + use super::*; + use serde_json::json; + + fn named(kind: &str, name: &str) -> serde_json::Value { + json!({"kind": kind, "name": name}) + } + + fn non_null(inner: serde_json::Value) -> serde_json::Value { + json!({"kind": "NON_NULL", "ofType": inner}) + } + + fn list(inner: serde_json::Value) -> serde_json::Value { + json!({"kind": "LIST", "ofType": inner}) + } + + /// A schema exercising each rendering path: an object with both an + /// undocumented and a documented argument list, an interface it implements, a + /// union, an enum holding a deprecated value, a `@oneOf` input object with + /// defaults, a custom scalar, and the built-in and meta types which listings + /// hide. + pub(crate) fn schema_fixture() -> Schema { + serde_json::from_value(json!({ + "queryType": {"name": "QueryRoot"}, + "mutationType": {"name": "MutationRoot"}, + "types": [ + // Hidden from listings unless `--all` is given. + {"kind": "SCALAR", "name": "String"}, + {"kind": "OBJECT", "name": "__Placeholder", "fields": []}, + + { + "kind": "SCALAR", + "name": "Id", + "description": "An opaque control-plane identifier.", + }, + { + "kind": "ENUM", + "name": "Capability", + "enumValues": [ + {"name": "read", "description": "May read."}, + { + "name": "admin", + "isDeprecated": true, + "deprecationReason": "Use \"write\" instead.", + }, + ], + }, + { + "kind": "INPUT_OBJECT", + "name": "PrivateLinkConfigInput", + "isOneOf": true, + "inputFields": [ + {"name": "aws", "type": named("INPUT_OBJECT", "AwsInput")}, + { + "name": "region", + "description": "Region of the link.", + "type": non_null(named("SCALAR", "String")), + "defaultValue": "\"us-east-1\"", + }, + { + "name": "zone", + "type": named("SCALAR", "String"), + "isDeprecated": true, + "deprecationReason": "Inferred from `region`.", + }, + ], + }, + {"kind": "INPUT_OBJECT", "name": "AwsInput", "inputFields": [ + {"name": "serviceName", "type": non_null(named("SCALAR", "String"))}, + ]}, + { + "kind": "INTERFACE", + "name": "Node", + "fields": [{"name": "id", "type": non_null(named("SCALAR", "Id"))}], + "possibleTypes": [named("OBJECT", "LiveSpec")], + }, + { + "kind": "OBJECT", + "name": "LiveSpec", + "description": "A live specification.\n\nOne per catalog name.", + "interfaces": [named("INTERFACE", "Node")], + "fields": [ + {"name": "id", "type": non_null(named("SCALAR", "Id"))}, + { + "name": "alerts", + "type": non_null(list(non_null(named("OBJECT", "Alert")))), + "args": [ + {"name": "first", "type": named("SCALAR", "Int")}, + {"name": "after", "type": named("SCALAR", "String")}, + ], + }, + { + "name": "prefixes", + "description": "Prefixes of this spec.", + "type": non_null(list(non_null(named("SCALAR", "String")))), + "args": [ + { + "name": "minCapability", + "description": "Least capability to report.", + "type": non_null(named("ENUM", "Capability")), + }, + { + "name": "first", + "type": named("SCALAR", "Int"), + "defaultValue": "10", + }, + ], + }, + { + "name": "oldName", + "type": named("SCALAR", "String"), + "isDeprecated": true, + "deprecationReason": "Use \"catalogName\" instead.", + }, + ], + }, + {"kind": "OBJECT", "name": "Alert", "fields": [ + {"name": "firedAt", "type": non_null(named("SCALAR", "String"))}, + ]}, + { + "kind": "UNION", + "name": "SpecOrAlert", + "possibleTypes": [named("OBJECT", "LiveSpec"), named("OBJECT", "Alert")], + }, + {"kind": "OBJECT", "name": "QueryRoot", "fields": [ + { + "name": "node", + "description": "Resolves any node by id.", + "type": named("INTERFACE", "Node"), + "args": [{"name": "id", "type": non_null(named("SCALAR", "Id"))}], + }, + ]}, + {"kind": "OBJECT", "name": "MutationRoot", "fields": [ + { + "name": "deleteLiveSpec", + "type": non_null(named("SCALAR", "Boolean")), + "args": [{"name": "id", "type": non_null(named("SCALAR", "Id"))}], + }, + ]}, + ], + "directives": [ + { + "name": "deprecated", + "description": "Marks an element as no longer supported.", + "locations": ["FIELD_DEFINITION", "ENUM_VALUE"], + "args": [{ + "name": "reason", + "type": named("SCALAR", "String"), + "defaultValue": "\"No longer supported\"", + }], + }, + {"name": "oneOf", "locations": ["INPUT_OBJECT"]}, + ], + })) + .expect("the fixture matches the introspection types") + } + + #[test] + fn test_render_schema() { + insta::assert_snapshot!(schema(&schema_fixture(), false)); + } + + #[test] + fn test_render_schema_all_includes_built_ins() { + let rendered = schema(&schema_fixture(), true); + + assert!(rendered.contains("scalar String\n")); + assert!(rendered.contains("type __Placeholder {")); + } + + #[test] + fn test_summarize() { + assert_eq!(summarize(None), ""); + assert_eq!(summarize(Some("\n\n first line \nsecond")), "first line"); + assert_eq!( + summarize(Some(&"x".repeat(200))), + format!("{}…", "x".repeat(96)) + ); + } +} diff --git a/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__render__test__render_schema.snap b/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__render__test__render_schema.snap new file mode 100644 index 00000000000..5cef216dd21 --- /dev/null +++ b/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__render__test__render_schema.snap @@ -0,0 +1,81 @@ +--- +source: crates/flowctl/src/raw/graphql/render.rs +expression: "schema(&schema_fixture(), false)" +--- +type Alert { + firedAt: String! +} + +input AwsInput { + serviceName: String! +} + +enum Capability { + """ + May read. + """ + read + admin @deprecated(reason: "Use \"write\" instead.") +} + +""" +An opaque control-plane identifier. +""" +scalar Id + +""" +A live specification. + +One per catalog name. +""" +type LiveSpec implements Node { + id: Id! + alerts(first: Int, after: String): [Alert!]! + """ + Prefixes of this spec. + """ + prefixes( + """ + Least capability to report. + """ + minCapability: Capability! + first: Int = 10 + ): [String!]! + oldName: String @deprecated(reason: "Use \"catalogName\" instead.") +} + +type MutationRoot { + deleteLiveSpec(id: Id!): Boolean! +} + +interface Node { + id: Id! +} + +input PrivateLinkConfigInput @oneOf { + aws: AwsInput + """ + Region of the link. + """ + region: String! = "us-east-1" + zone: String @deprecated(reason: "Inferred from `region`.") +} + +type QueryRoot { + """ + Resolves any node by id. + """ + node(id: Id!): Node +} + +union SpecOrAlert = LiveSpec | Alert + +""" +Marks an element as no longer supported. +""" +directive @deprecated(reason: String = "No longer supported") on FIELD_DEFINITION | ENUM_VALUE +directive @oneOf on INPUT_OBJECT +schema { + query: QueryRoot + mutation: MutationRoot +} diff --git a/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__test__build_variables-2.snap b/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__test__build_variables-2.snap new file mode 100644 index 00000000000..16b22e0ed62 --- /dev/null +++ b/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__test__build_variables-2.snap @@ -0,0 +1,8 @@ +--- +source: crates/flowctl/src/raw/graphql/mod.rs +expression: "build_variables(Some(r#\"{\"first\": 5, \"prefix\": \"wileyCo/\"}\"#),\n&[var(\"first\", \"10\")]).unwrap()" +--- +{ + "first": 10, + "prefix": "wileyCo/" +} diff --git a/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__test__build_variables.snap b/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__test__build_variables.snap new file mode 100644 index 00000000000..198c0eac951 --- /dev/null +++ b/crates/flowctl/src/raw/graphql/snapshots/flowctl__raw__graphql__test__build_variables.snap @@ -0,0 +1,15 @@ +--- +source: crates/flowctl/src/raw/graphql/mod.rs +expression: "build_variables(None, &vars).unwrap()" +--- +{ + "closed": true, + "filter": { + "catalogPrefix": { + "startsWith": "acmeCo/" + } + }, + "first": 10, + "prefix": "acmeCo/", + "quoted": "10" +} diff --git a/crates/flowctl/src/raw/mod.rs b/crates/flowctl/src/raw/mod.rs index 3f5321e4186..d404c29adb6 100644 --- a/crates/flowctl/src/raw/mod.rs +++ b/crates/flowctl/src/raw/mod.rs @@ -13,6 +13,7 @@ use tables::CatalogResolver; mod alerts; mod discover; +mod graphql; mod materialize_fixture; mod oauth; mod preview_next; @@ -51,6 +52,14 @@ pub enum Command { Rpc(Rpc), /// Issue a custom table update request to the API. Update(Update), + /// Explore the control-plane GraphQL API, and run queries and mutations + /// against it. + /// + /// The GraphQL API is the successor of the PostgREST API which `get`, `rpc`, + /// and `update` speak to. Use `graphql operations` to see what it serves, + /// `graphql describe` to see the shape of a type, and `graphql exec` to run + /// an operation. + Graphql(graphql::Graphql), /// Perform a configured build of catalog sources. Build(Build), /// Bundle catalog sources into a flattened and inlined catalog. @@ -215,6 +224,7 @@ impl Advanced { Command::Get(get) => do_get(ctx, get).await, Command::Update(update) => do_update(ctx, update).await, Command::Rpc(rpc) => do_rpc(ctx, rpc).await, + Command::Graphql(args) => graphql::do_graphql(ctx, args).await, Command::Build(build) => do_build(ctx, build).await, Command::Bundle(bundle) => do_bundle(ctx, bundle).await, Command::Combine(combine) => do_combine(ctx, combine).await,