diff --git a/.sqlx/query-1ceb98ba09ec9a70b74e88bd057a278ecdf1c34cf831c51a0ac4d0002a4bd3ff.json b/.sqlx/query-1ceb98ba09ec9a70b74e88bd057a278ecdf1c34cf831c51a0ac4d0002a4bd3ff.json new file mode 100644 index 00000000000..afe8201a7e6 --- /dev/null +++ b/.sqlx/query-1ceb98ba09ec9a70b74e88bd057a278ecdf1c34cf831c51a0ac4d0002a4bd3ff.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n job_status->>'type' = 'success' AS \"processed!\",\n coalesce(protocol = 'capture', false) AS \"is_capture!\"\n FROM connector_tags\n WHERE id = $1::flowid\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "processed!", + "type_info": "Bool" + }, + { + "ordinal": 1, + "name": "is_capture!", + "type_info": "Bool" + } + ], + "parameters": { + "Left": [ + { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + } + ] + }, + "nullable": [ + null, + null + ] + }, + "hash": "1ceb98ba09ec9a70b74e88bd057a278ecdf1c34cf831c51a0ac4d0002a4bd3ff" +} diff --git a/.sqlx/query-4a18c6a1e1857b3c7d2c87971e29d6e653dcedb3524c4a2d364215b551b8a63e.json b/.sqlx/query-4a18c6a1e1857b3c7d2c87971e29d6e653dcedb3524c4a2d364215b551b8a63e.json new file mode 100644 index 00000000000..f14c19b2f3d --- /dev/null +++ b/.sqlx/query-4a18c6a1e1857b3c7d2c87971e29d6e653dcedb3524c4a2d364215b551b8a63e.json @@ -0,0 +1,90 @@ +{ + "db_name": "PostgreSQL", + "query": "\n SELECT\n d.id AS \"id: Id\",\n d.draft_id AS \"draft_id: Id\",\n d.capture_name AS \"capture_name: models::Capture\",\n d.connector_tag_id AS \"connector_tag_id: Id\",\n d.data_plane_name AS \"data_plane_name: models::Name\",\n d.update_only,\n d.detail,\n d.logs_token,\n d.job_status->>'type' AS \"status!: String\",\n d.created_at,\n d.updated_at\n FROM discovers d\n JOIN drafts ON drafts.id = d.draft_id\n WHERE d.id = $1::flowid AND drafts.user_id = $2\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "draft_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 2, + "name": "capture_name: models::Capture", + "type_info": "Text" + }, + { + "ordinal": 3, + "name": "connector_tag_id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 4, + "name": "data_plane_name: models::Name", + "type_info": "Text" + }, + { + "ordinal": 5, + "name": "update_only", + "type_info": "Bool" + }, + { + "ordinal": 6, + "name": "detail", + "type_info": "Text" + }, + { + "ordinal": 7, + "name": "logs_token", + "type_info": "Uuid" + }, + { + "ordinal": 8, + "name": "status!: String", + "type_info": "Text" + }, + { + "ordinal": 9, + "name": "created_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 10, + "name": "updated_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + }, + "Uuid" + ] + }, + "nullable": [ + false, + false, + false, + false, + false, + false, + true, + false, + null, + false, + false + ] + }, + "hash": "4a18c6a1e1857b3c7d2c87971e29d6e653dcedb3524c4a2d364215b551b8a63e" +} diff --git a/.sqlx/query-5ff339c03387e6c36eff5d3851fef4f9211add9ad0d3761a11782d05216536b1.json b/.sqlx/query-5ff339c03387e6c36eff5d3851fef4f9211add9ad0d3761a11782d05216536b1.json new file mode 100644 index 00000000000..22f5abb803c --- /dev/null +++ b/.sqlx/query-5ff339c03387e6c36eff5d3851fef4f9211add9ad0d3761a11782d05216536b1.json @@ -0,0 +1,67 @@ +{ + "db_name": "PostgreSQL", + "query": "\n INSERT INTO discovers (\n capture_name, connector_tag_id, draft_id, endpoint_config,\n update_only, data_plane_name, detail\n )\n VALUES ($1::text::catalog_name, $2::flowid, $3::flowid, $4, $5, $6, $7)\n RETURNING id AS \"id: Id\", logs_token, created_at, updated_at\n ", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: Id", + "type_info": "Macaddr8" + }, + { + "ordinal": 1, + "name": "logs_token", + "type_info": "Uuid" + }, + { + "ordinal": 2, + "name": "created_at", + "type_info": "Timestamptz" + }, + { + "ordinal": 3, + "name": "updated_at", + "type_info": "Timestamptz" + } + ], + "parameters": { + "Left": [ + "Text", + { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + }, + { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + }, + { + "Custom": { + "name": "json_obj", + "kind": { + "Domain": "Json" + } + } + }, + "Bool", + "Text", + "Text" + ] + }, + "nullable": [ + false, + false, + false, + false + ] + }, + "hash": "5ff339c03387e6c36eff5d3851fef4f9211add9ad0d3761a11782d05216536b1" +} diff --git a/.sqlx/query-8471b76108c64dc8431fc175576571e81d090d2863ebf5d9248c396ce534a136.json b/.sqlx/query-8471b76108c64dc8431fc175576571e81d090d2863ebf5d9248c396ce534a136.json new file mode 100644 index 00000000000..2a4dbb39b26 --- /dev/null +++ b/.sqlx/query-8471b76108c64dc8431fc175576571e81d090d2863ebf5d9248c396ce534a136.json @@ -0,0 +1,23 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO drafts (user_id, detail) VALUES ($1, $2) RETURNING id AS \"id: Id\"", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: Id", + "type_info": "Macaddr8" + } + ], + "parameters": { + "Left": [ + "Uuid", + "Text" + ] + }, + "nullable": [ + false + ] + }, + "hash": "8471b76108c64dc8431fc175576571e81d090d2863ebf5d9248c396ce534a136" +} diff --git a/.sqlx/query-ab9d0f7d5cc6d68c588edaa6bb8a1fcf891f0e6927e94f2bda56d775d74a139f.json b/.sqlx/query-ab9d0f7d5cc6d68c588edaa6bb8a1fcf891f0e6927e94f2bda56d775d74a139f.json new file mode 100644 index 00000000000..8683e9ada24 --- /dev/null +++ b/.sqlx/query-ab9d0f7d5cc6d68c588edaa6bb8a1fcf891f0e6927e94f2bda56d775d74a139f.json @@ -0,0 +1,30 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT id AS \"id: Id\" FROM drafts WHERE id = $1::flowid AND user_id = $2", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id: Id", + "type_info": "Macaddr8" + } + ], + "parameters": { + "Left": [ + { + "Custom": { + "name": "flowid", + "kind": { + "Domain": "Macaddr8" + } + } + }, + "Uuid" + ] + }, + "nullable": [ + false + ] + }, + "hash": "ab9d0f7d5cc6d68c588edaa6bb8a1fcf891f0e6927e94f2bda56d775d74a139f" +} diff --git a/crates/agent/src/discovers.rs b/crates/agent/src/discovers.rs index 23c27f5001c..47bd2e94f21 100644 --- a/crates/agent/src/discovers.rs +++ b/crates/agent/src/discovers.rs @@ -6,39 +6,7 @@ use control_plane_api::{ proxy_connectors::DiscoverConnectors, }; use models::Id; -use serde::{Deserialize, Serialize}; - -/// JobStatus is the possible outcomes of a handled discover operation. -#[derive(Debug, Deserialize, Serialize)] -#[serde(rename_all = "camelCase", tag = "type")] -pub enum JobStatus { - Queued, - WrongProtocol, - TagFailed, - ImageForbidden, - PullFailed, - DiscoverFailed, - MergeFailed, - Success { - #[serde(default, skip_serializing_if = "Option::is_none")] - publication_id: Option, - #[serde(default, skip_serializing_if = "std::ops::Not::not")] - specs_unchanged: bool, - }, - DeprecatedBackground, - NoDataPlane, - NotAuthorized, -} - -impl JobStatus { - #[cfg(test)] - pub fn is_success(&self) -> bool { - match self { - JobStatus::Success { .. } => true, - _ => false, - } - } -} +use models::discovers::JobStatus; type ProcessResult = Result>; @@ -403,16 +371,6 @@ mod test { use models::Id; use uuid::Uuid; - // `discovers.job_status` is a UI-facing contract: pin the wire tag. - #[test] - fn test_job_status_not_authorized_serde() { - let status = serde_json::to_value(super::JobStatus::NotAuthorized).unwrap(); - assert_eq!(serde_json::json!({"type": "notAuthorized"}), status); - - let round: super::JobStatus = serde_json::from_value(status).unwrap(); - assert!(matches!(round, super::JobStatus::NotAuthorized)); - } - #[tokio::test] async fn test_prepare_discover() { let harness = diff --git a/crates/agent/src/integration_tests/graphql/mutations/discovers.rs b/crates/agent/src/integration_tests/graphql/mutations/discovers.rs new file mode 100644 index 00000000000..4b4d8cfa453 --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/mutations/discovers.rs @@ -0,0 +1,256 @@ +//! End-to-end coverage of the GraphQL discover surface: `createDiscover` runs +//! against the real schema in-process, the real `DiscoverExecutor` polls the +//! task the row's trigger enqueued, and `discover(id)` reports the outcome. + +use crate::integration_tests::harness::{TestHarness, draft_catalog}; +use crate::integration_tests::spec_fixture; +use proto_flow::capture::response::{Discovered, discovered::Binding}; +use serde_json::json; +use sqlx::types::Uuid; + +const CREATE_DISCOVER: &str = r#" +mutation($captureName: Capture!, $connectorTagId: Id!, $draftId: Id, $endpointConfig: JSON!) { + createDiscover( + captureName: $captureName + connectorTagId: $connectorTagId + dataPlaneName: "ops/dp/public/test" + draftId: $draftId + endpointConfig: $endpointConfig + ) { + id + draftId + status + } +} +"#; + +const CAPTURE: &str = "aliceCo/e2e/capture"; + +const DISCOVER_STATUS: &str = r#" +query($id: Id!) { + discover(id: $id) { status } +} +"#; + +/// A discover response naming one collection per `names` entry, all keyed on `/id`. +fn discovered(names: &[&str]) -> Discovered { + Discovered { + bindings: names + .iter() + .map(|name| Binding { + recommended_name: name.to_string(), + document_schema_json: json!({ + "type": "object", + "properties": { "id": { "type": "string" } }, + "required": ["id"], + }) + .to_string() + .into(), + resource_config_json: json!({ "id": name }).to_string().into(), + key: vec!["/id".to_string()], + disable: false, + resource_path: Vec::new(), + is_fallback_key: false, + }) + .collect(), + } +} + +/// The connector tag id a client would obtain before calling `createDiscover`. +async fn source_test_tag_id(harness: &mut TestHarness, user_id: Uuid) -> String { + let response: serde_json::Value = harness + .execute_graphql_query( + user_id, + r#"query { connector(imageName: "source/test") { spec(imageTag: ":test") { id } } }"#, + &json!({}), + ) + .await + .unwrap(); + response["connector"]["spec"]["id"] + .as_str() + .unwrap_or_else(|| panic!("no source/test:test spec in {response}")) + .to_string() +} + +async fn create_discover( + harness: &mut TestHarness, + user_id: Uuid, + capture_name: &str, + connector_tag_id: &str, + draft_id: Option, +) -> anyhow::Result { + let response: serde_json::Value = harness + .execute_graphql_query( + user_id, + CREATE_DISCOVER, + &json!({ + "captureName": capture_name, + "connectorTagId": connector_tag_id, + "draftId": draft_id, + "endpointConfig": { "tail": "shake" }, + }), + ) + .await?; + Ok(response["createDiscover"].clone()) +} + +async fn discover_status(harness: &mut TestHarness, user_id: Uuid, id: models::Id) -> String { + let response: serde_json::Value = harness + .execute_graphql_query(user_id, DISCOVER_STATUS, &json!({ "id": id })) + .await + .unwrap(); + response["discover"]["status"].as_str().unwrap().to_string() +} + +fn parse_id(value: &serde_json::Value) -> models::Id { + models::Id::from_hex(value.as_str().unwrap()).unwrap() +} + +fn assert_success(result: &crate::integration_tests::harness::UserDiscoverResult) { + assert!( + result.job_status.is_success(), + "expected success, got: {:?} with errors {:?}", + result.job_status, + result.errors + ); +} + +#[tokio::test] +async fn test_discover_end_to_end_with_new_draft() { + let mut harness = TestHarness::init("graphql_discover_new_draft").await; + let alice = harness.setup_tenant("aliceCo").await; + let tag_id = source_test_tag_id(&mut harness, alice).await; + harness.discover_handler.connectors.mock_discover( + CAPTURE, + Ok((spec_fixture(), discovered(&["acorns", "walnuts"]))), + ); + + let created = create_discover(&mut harness, alice, CAPTURE, &tag_id, None) + .await + .unwrap(); + assert_eq!("QUEUED", created["status"]); + let discover_id = parse_id(&created["id"]); + let draft_id = parse_id(&created["draftId"]); + + let result = harness.run_queued_discover(discover_id).await; + assert_success(&result); + assert_eq!( + "SUCCESS", + discover_status(&mut harness, alice, discover_id).await + ); + + // The mutation created the draft for the caller, and the executor filled + // it with the capture and its collections. + let owner: Uuid = sqlx::query_scalar("SELECT user_id FROM drafts WHERE id = $1") + .bind(draft_id) + .fetch_one(&harness.pool) + .await + .unwrap(); + assert_eq!(alice, owner); + insta::assert_debug_snapshot!("new-draft", result.draft); +} + +#[tokio::test] +async fn test_discover_end_to_end_into_existing_draft() { + let mut harness = TestHarness::init("graphql_discover_existing_draft").await; + let alice = harness.setup_tenant("aliceCo").await; + let tag_id = source_test_tag_id(&mut harness, alice).await; + + // The draft already holds the capture with one binding and an unrelated + // collection. A re-discover must keep both and merge the new binding. + let draft_id = harness + .create_draft( + alice, + "pre-existing work", + draft_catalog(json!({ + "captures": { + CAPTURE: { + "endpoint": { "connector": { "image": "source/test:test", "config": {} } }, + "bindings": [ + { + "resource": { "id": "acorns", "kept": "drafted resource config" }, + "target": "aliceCo/e2e/acorns" + } + ] + } + }, + "collections": { + "aliceCo/e2e/unrelated": { + "schema": { "type": "object", "properties": { "id": { "type": "string" } } }, + "key": ["/id"] + } + } + })), + ) + .await; + harness.discover_handler.connectors.mock_discover( + CAPTURE, + Ok((spec_fixture(), discovered(&["acorns", "walnuts"]))), + ); + + let created = create_discover(&mut harness, alice, CAPTURE, &tag_id, Some(draft_id)) + .await + .unwrap(); + assert_eq!(draft_id, parse_id(&created["draftId"])); + let discover_id = parse_id(&created["id"]); + + let result = harness.run_queued_discover(discover_id).await; + assert_success(&result); + assert_eq!( + "SUCCESS", + discover_status(&mut harness, alice, discover_id).await + ); + insta::assert_debug_snapshot!("existing-draft", result.draft); +} + +#[tokio::test] +async fn test_discover_end_to_end_connector_failure() { + let mut harness = TestHarness::init("graphql_discover_connector_failure").await; + let alice = harness.setup_tenant("aliceCo").await; + let tag_id = source_test_tag_id(&mut harness, alice).await; + harness.discover_handler.connectors.mock_discover( + CAPTURE, + Err("the connector could not reach the endpoint".to_string()), + ); + + let created = create_discover(&mut harness, alice, CAPTURE, &tag_id, None) + .await + .unwrap(); + let discover_id = parse_id(&created["id"]); + + let result = harness.run_queued_discover(discover_id).await; + assert_eq!( + "DISCOVER_FAILED", + discover_status(&mut harness, alice, discover_id).await + ); + insta::assert_debug_snapshot!("connector-failure-errors", result.errors); +} + +#[tokio::test] +async fn test_discover_end_to_end_unauthorized_capture() { + let mut harness = TestHarness::init("graphql_discover_unauthorized").await; + let alice = harness.setup_tenant("aliceCo").await; + let tag_id = source_test_tag_id(&mut harness, alice).await; + + // Alice holds nothing under bobCo/, so the mutation is rejected before a + // row is written and there is no task for the executor to run. + let err = create_discover(&mut harness, alice, "bobCo/e2e/capture", &tag_id, None) + .await + .unwrap_err() + .to_string(); + assert!( + err.contains("PermissionDenied") && err.contains("SpecEdit"), + "unexpected error: {err}" + ); + assert_eq!( + None, + harness + .run_automation_task(automations::task_types::DISCOVERS) + .await + ); + let rows: i64 = sqlx::query_scalar("SELECT count(*) FROM discovers") + .fetch_one(&harness.pool) + .await + .unwrap(); + assert_eq!(0, rows); +} diff --git a/crates/agent/src/integration_tests/graphql/mutations/mod.rs b/crates/agent/src/integration_tests/graphql/mutations/mod.rs index 5fd1f0eb962..166b0773d4a 100644 --- a/crates/agent/src/integration_tests/graphql/mutations/mod.rs +++ b/crates/agent/src/integration_tests/graphql/mutations/mod.rs @@ -1 +1,2 @@ +mod discovers; mod storage_mappings; diff --git a/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__connector-failure-errors.snap b/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__connector-failure-errors.snap new file mode 100644 index 00000000000..f6851860c42 --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__connector-failure-errors.snap @@ -0,0 +1,10 @@ +--- +source: crates/agent/src/integration_tests/graphql/mutations/discovers.rs +expression: result.errors +--- +[ + ( + "flow://capture/aliceCo/e2e/capture", + "the connector could not reach the endpoint", + ), +] diff --git a/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__existing-draft.snap b/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__existing-draft.snap new file mode 100644 index 00000000000..b46a00dbc32 --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__existing-draft.snap @@ -0,0 +1,70 @@ +--- +source: crates/agent/src/integration_tests/graphql/mutations/discovers.rs +expression: result.draft +--- +DraftCatalog { + captures: [ + DraftCapture { + capture: aliceCo/e2e/capture, + scope: flow://capture/aliceCo/e2e/capture, + expect_pub_id: NULL, + model: { + "endpoint": { + "connector": { + "image": "source/test:test", + "config": {"tail":"shake"} + } + }, + "bindings": [ + { + "resource": {"id":"acorns","kept":"drafted resource config"}, + "target": "aliceCo/e2e/acorns" + }, + { + "resource": {"id":"walnuts"}, + "target": "aliceCo/e2e/walnuts" + } + ] + }, + is_touch: 0, + }, + ], + collections: [ + DraftCollection { + collection: aliceCo/e2e/acorns, + scope: flow://collection/aliceCo/e2e/acorns, + expect_pub_id: "0000000000000000", + model: { + "schema": {"$defs":{"flow://connector-schema":{"$id":"flow://connector-schema","properties":{"id":{"type":"string"}},"required":["id"],"type":"object"}},"$ref":"flow://connector-schema"}, + "key": [ + "/id" + ] + }, + is_touch: 0, + }, + DraftCollection { + collection: aliceCo/e2e/unrelated, + scope: flow://collection/aliceCo/e2e/unrelated, + expect_pub_id: NULL, + model: { + "schema": {"properties":{"id":{"type":"string"}},"type":"object"}, + "key": [ + "/id" + ] + }, + is_touch: 0, + }, + DraftCollection { + collection: aliceCo/e2e/walnuts, + scope: flow://collection/aliceCo/e2e/walnuts, + expect_pub_id: "0000000000000000", + model: { + "schema": {"$defs":{"flow://connector-schema":{"$id":"flow://connector-schema","properties":{"id":{"type":"string"}},"required":["id"],"type":"object"}},"$ref":"flow://connector-schema"}, + "key": [ + "/id" + ] + }, + is_touch: 0, + }, + ], +} diff --git a/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__new-draft.snap b/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__new-draft.snap new file mode 100644 index 00000000000..8835d852854 --- /dev/null +++ b/crates/agent/src/integration_tests/graphql/mutations/snapshots/agent__integration_tests__graphql__mutations__discovers__new-draft.snap @@ -0,0 +1,62 @@ +--- +source: crates/agent/src/integration_tests/graphql/mutations/discovers.rs +expression: result.draft +--- +DraftCatalog { + captures: [ + DraftCapture { + capture: aliceCo/e2e/capture, + scope: flow://capture/aliceCo/e2e/capture, + expect_pub_id: "0000000000000000", + model: { + "autoDiscover": { + "addNewBindings": true, + "evolveIncompatibleCollections": true + }, + "endpoint": { + "connector": { + "image": "source/test:test", + "config": {"tail":"shake"} + } + }, + "bindings": [ + { + "resource": {"id":"acorns"}, + "target": "aliceCo/e2e/acorns" + }, + { + "resource": {"id":"walnuts"}, + "target": "aliceCo/e2e/walnuts" + } + ] + }, + is_touch: 0, + }, + ], + collections: [ + DraftCollection { + collection: aliceCo/e2e/acorns, + scope: flow://collection/aliceCo/e2e/acorns, + expect_pub_id: "0000000000000000", + model: { + "schema": {"$defs":{"flow://connector-schema":{"$id":"flow://connector-schema","properties":{"id":{"type":"string"}},"required":["id"],"type":"object"}},"$ref":"flow://connector-schema"}, + "key": [ + "/id" + ] + }, + is_touch: 0, + }, + DraftCollection { + collection: aliceCo/e2e/walnuts, + scope: flow://collection/aliceCo/e2e/walnuts, + expect_pub_id: "0000000000000000", + model: { + "schema": {"$defs":{"flow://connector-schema":{"$id":"flow://connector-schema","properties":{"id":{"type":"string"}},"required":["id"],"type":"object"}},"$ref":"flow://connector-schema"}, + "key": [ + "/id" + ] + }, + is_touch: 0, + }, + ], +} diff --git a/crates/agent/src/integration_tests/harness.rs b/crates/agent/src/integration_tests/harness.rs index 2948e93393b..19729eb08ab 100644 --- a/crates/agent/src/integration_tests/harness.rs +++ b/crates/agent/src/integration_tests/harness.rs @@ -9,7 +9,6 @@ use crate::publications::PublicationsExecutor; use crate::{ controllers::ControllerState, controlplane::{ConnectorSpec, ControlPlane, PGControlPlane}, - discovers, }; use anyhow::Context; use chrono::{DateTime, Utc}; @@ -123,7 +122,7 @@ pub struct ScenarioResult { } pub struct UserDiscoverResult { - pub job_status: crate::discovers::JobStatus, + pub job_status: models::discovers::JobStatus, pub draft: tables::DraftCatalog, pub errors: Vec<(String, String)>, } @@ -133,7 +132,7 @@ impl UserDiscoverResult { let discover = sqlx::query!( r#"select draft_id as "draft_id: Id", - job_status as "job_status: TextJson" + job_status as "job_status: TextJson" from discovers where id = $1;"#, discover_id as Id, diff --git a/crates/agent/src/integration_tests/user_discovers.rs b/crates/agent/src/integration_tests/user_discovers.rs index 1b5eddbf9bf..3f5c488515a 100644 --- a/crates/agent/src/integration_tests/user_discovers.rs +++ b/crates/agent/src/integration_tests/user_discovers.rs @@ -402,7 +402,7 @@ async fn test_discover_not_authorized_capture() { assert!( matches!( result.job_status, - crate::discovers::JobStatus::NotAuthorized + models::discovers::JobStatus::NotAuthorized ), "{case}: expected NotAuthorized, got: {:?}", result.job_status @@ -485,7 +485,7 @@ async fn test_discover_capture_requires_spec_edit() { assert!( matches!( result.job_status, - crate::discovers::JobStatus::NotAuthorized + models::discovers::JobStatus::NotAuthorized ), "{capability:?}: expected NotAuthorized, got: {:?}", result.job_status @@ -528,7 +528,7 @@ async fn test_discover_not_authorized_wins_over_missing_plane() { assert!( matches!( result.job_status, - crate::discovers::JobStatus::NotAuthorized + models::discovers::JobStatus::NotAuthorized ), "expected NotAuthorized, got: {:?}", result.job_status @@ -589,7 +589,7 @@ async fn test_discover_no_data_plane() { let result = harness.run_queued_discover(discover_id).await; assert!( - matches!(result.job_status, crate::discovers::JobStatus::NoDataPlane), + matches!(result.job_status, models::discovers::JobStatus::NoDataPlane), "{case}: expected NoDataPlane, got: {:?}", result.job_status ); diff --git a/crates/control-plane-api/src/server/public/graphql/discovers.rs b/crates/control-plane-api/src/server/public/graphql/discovers.rs new file mode 100644 index 00000000000..e8ac14e245e --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/discovers.rs @@ -0,0 +1,924 @@ +//! GraphQL surface for user-initiated connector discovers. +//! +//! A discover is visible to exactly one person: the owner of the draft it +//! merges into. `discovers` carries no `user_id` of its own, and this API reads +//! Postgres on a privileged connection where row-level security does not +//! apply, so the join to `drafts` on the caller's id *is* the access check. +//! Omitting it would expose every discover, including its `logsToken` bearer +//! credential, to any caller holding an id. + +use async_graphql::Context; +use models::Id; + +/// A user-initiated connector discovery operation. +#[derive(Debug, Clone, async_graphql::SimpleObject)] +pub struct Discover { + pub id: Id, + /// The draft that discovered specs are merged into. + pub draft_id: Id, + /// Catalog name of the capture being discovered. + pub capture_name: models::Capture, + pub connector_tag_id: Id, + pub data_plane_name: models::Name, + pub update_only: bool, + pub detail: Option, + /// Bearer token for reading this operation's logs. + pub logs_token: uuid::Uuid, + /// Current outcome of the operation. `QUEUED` until the server has + /// processed it. + pub status: models::discovers::DiscoverStatusType, + pub created_at: chrono::DateTime, + pub updated_at: chrono::DateTime, +} + +#[derive(Debug, Default)] +pub struct DiscoversQuery; + +#[derive(Debug, Default)] +pub struct DiscoversMutation; + +#[async_graphql::Object] +impl DiscoversMutation { + /// Queue a connector discovery for `captureName`, merging the discovered + /// bindings and collections into a draft. + /// + /// When `draftId` is given, that draft is used and must be owned by the + /// caller. When it is omitted, a new draft owned by the caller is created + /// and its id is returned as `Discover.draftId`. + /// + /// The caller must hold `SpecEdit` on `captureName` and read on + /// `dataPlaneName`. The connector tag must exist, have a successfully + /// processed spec, and be a capture connector. Poll `discover(id)` for the + /// outcome; the merged specs are in the draft. + /// + /// Note that a legacy `write` grant does not convey `SpecEdit`: users who + /// can read and append to a prefix's collections still cannot discover + /// under it. Legacy `admin` on a prefix covering the capture satisfies + /// both requirements in a normally provisioned tenant. The `Editor` bundle + /// conveys `SpecEdit` but on its own does not reach the public data + /// planes through the tenant's role grant. + async fn create_discover( + &self, + ctx: &Context<'_>, + #[graphql(desc = "Catalog name of the capture to discover. It need not exist yet.")] + capture_name: models::Capture, + #[graphql( + desc = "Id of the connector tag to run, as returned by `connector.spec(imageTag).id`." + )] + connector_tag_id: Id, + #[graphql( + desc = "Data plane in which to run the connector. The caller must hold read on it." + )] + data_plane_name: models::Name, + #[graphql( + desc = "Endpoint configuration passed to the connector, stored verbatim. Object \ + field order is preserved so that sops-encrypted configurations remain \ + decryptable. Encryption is the caller's responsibility." + )] + endpoint_config: async_graphql::Json, + #[graphql( + desc = "The draft that discovered specs are merged into. Must be owned by the \ + caller. When omitted, a new draft is created for this discover." + )] + draft_id: Option, + #[graphql( + default = false, + desc = "When true, existing bindings are refreshed and newly discovered bindings \ + are added in a disabled state." + )] + update_only: bool, + #[graphql( + desc = "Optional description, recorded on the discover and on a draft this \ + mutation creates." + )] + detail: Option, + ) -> async_graphql::Result { + let env = ctx.data::()?; + let claims = env.claims()?; + + if let Err(err) = validator::Validate::validate(&capture_name) { + return Err(async_graphql::Error::new(format!( + "invalid capture name: {err}" + ))); + } + + // The executor re-checks both of these against its own pinned + // Snapshot, but a denial there surfaces only as a terminal job status + // after the job has been queued. Gating here makes it a synchronous + // error instead. + super::verify_authorization(env, &capture_name, models::authz::Capability::SpecEdit) + .await?; + + // Resolving the plane consults what exists in the Snapshot, so an + // unauthorized plane and a missing one are deliberately + // indistinguishable: a terminal denial becomes the same "not found" + // as an absent plane, while a provisional denial against a stale + // Snapshot still takes the standard refresh-and-retry path. + let snapshot = env.snapshot(); + let policy_result = crate::server::evaluate_names_authorization( + snapshot, + claims, + models::Capability::Read, + [data_plane_name.as_str()], + ); + let may_read_plane = match env.authorization_outcome(policy_result).await { + Ok(_) => true, + Err(retry @ crate::ApiError::AuthZRetry(_)) => return Err(retry.into()), + Err(_) => false, + }; + if !may_read_plane + || snapshot + .data_plane_by_catalog_name(&data_plane_name) + .is_none() + { + return Err(async_graphql::Error::new(format!( + "data plane {data_plane_name} was not found" + ))); + } + + // `protocol` is null until the tag's spec job succeeds, and a null + // comparison would fail the non-null decode rather than read as false. + let Some(tag) = sqlx::query!( + r#" + SELECT + job_status->>'type' = 'success' AS "processed!", + coalesce(protocol = 'capture', false) AS "is_capture!" + FROM connector_tags + WHERE id = $1::flowid + "#, + connector_tag_id as Id, + ) + .fetch_optional(&env.pg_pool) + .await? + else { + return Err(async_graphql::Error::new("connector tag not found")); + }; + if !tag.processed { + return Err(async_graphql::Error::new(format!( + "connector tag {connector_tag_id} has not been successfully processed" + ))); + } + if !tag.is_capture { + return Err(async_graphql::Error::new(format!( + "connector tag {connector_tag_id} is not a capture connector" + ))); + } + + let mut txn = env.pg_pool.begin().await?; + + // A given draft must belong to the caller; missing and not-owned are + // the same "not found" so ids cannot probe other users' drafts. + let draft_id = match draft_id { + Some(draft_id) => sqlx::query_scalar!( + r#"SELECT id AS "id: Id" FROM drafts WHERE id = $1::flowid AND user_id = $2"#, + draft_id as Id, + claims.sub, + ) + .fetch_optional(&mut *txn) + .await? + .ok_or_else(|| async_graphql::Error::new("draft not found"))?, + None => sqlx::query_scalar!( + r#"INSERT INTO drafts (user_id, detail) VALUES ($1, $2) RETURNING id AS "id: Id""#, + claims.sub, + detail.as_deref(), + ) + .fetch_one(&mut *txn) + .await?, + }; + + // `async_graphql::Value` keeps object fields in input order, and JSON + // (not JSONB) storage keeps them on the way in, so the stored + // configuration has the caller's field order (whitespace is not kept). + let endpoint_config = + models::RawValue::from_string(serde_json::to_string(&endpoint_config.0)?)?; + + // The `create_discover_task` trigger enqueues the executor's task. + let row = sqlx::query!( + r#" + INSERT INTO discovers ( + capture_name, connector_tag_id, draft_id, endpoint_config, + update_only, data_plane_name, detail + ) + VALUES ($1::text::catalog_name, $2::flowid, $3::flowid, $4, $5, $6, $7) + RETURNING id AS "id: Id", logs_token, created_at, updated_at + "#, + capture_name.as_str(), + connector_tag_id as Id, + draft_id as Id, + crate::TextJson(endpoint_config) as crate::TextJson, + update_only, + data_plane_name.as_str(), + detail.as_deref(), + ) + .fetch_one(&mut *txn) + .await?; + + txn.commit().await?; + + tracing::info!( + discover_id = %row.id, + %draft_id, + %capture_name, + %connector_tag_id, + %data_plane_name, + user_id = %claims.sub, + "queued discover" + ); + + Ok(Discover { + id: row.id, + draft_id, + capture_name, + connector_tag_id, + data_plane_name, + update_only, + detail, + logs_token: row.logs_token, + status: models::discovers::DiscoverStatusType::Queued, + created_at: row.created_at, + updated_at: row.updated_at, + }) + } +} + +#[async_graphql::Object] +impl DiscoversQuery { + /// Fetch a single discover by id. Returns null if no such discover exists + /// or the caller does not own its draft; the two cases are + /// indistinguishable. + async fn discover(&self, ctx: &Context<'_>, id: Id) -> async_graphql::Result> { + let env = ctx.data::()?; + let claims = env.claims()?; + + let Some(row) = sqlx::query!( + r#" + SELECT + d.id AS "id: Id", + d.draft_id AS "draft_id: Id", + d.capture_name AS "capture_name: models::Capture", + d.connector_tag_id AS "connector_tag_id: Id", + d.data_plane_name AS "data_plane_name: models::Name", + d.update_only, + d.detail, + d.logs_token, + d.job_status->>'type' AS "status!: String", + d.created_at, + d.updated_at + FROM discovers d + JOIN drafts ON drafts.id = d.draft_id + WHERE d.id = $1::flowid AND drafts.user_id = $2 + "#, + id as Id, + claims.sub, + ) + .fetch_optional(&env.pg_pool) + .await? + else { + return Ok(None); + }; + + // The stored tag is the camelCase serde name of DiscoverStatusType. + let status = serde_json::from_value(serde_json::Value::String(row.status))?; + + Ok(Some(Discover { + id: row.id, + draft_id: row.draft_id, + capture_name: row.capture_name, + connector_tag_id: row.connector_tag_id, + data_plane_name: row.data_plane_name, + update_only: row.update_only, + detail: row.detail, + logs_token: row.logs_token, + status, + created_at: row.created_at, + updated_at: row.updated_at, + })) + } +} + +#[cfg(test)] +mod test { + use crate::test_server; + + const DISCOVER_QUERY: &str = r#" + query($id: Id!) { + discover(id: $id) { + id + draftId + captureName + connectorTagId + dataPlaneName + updateOnly + detail + logsToken + status + createdAt + updatedAt + } + }"#; + + const ALICE: uuid::Uuid = uuid::Uuid::from_bytes([0x11; 16]); + const BOB: uuid::Uuid = uuid::Uuid::from_bytes([0x22; 16]); + const SERVICE_ACCOUNT: uuid::Uuid = uuid::Uuid::from_bytes([0x33; 16]); + const CAROL: uuid::Uuid = uuid::Uuid::from_bytes([0x44; 16]); + const SOURCE_TAG: &str = "6666666600000000"; + // Deliberately unsorted so that a stored copy proves field order survived. + const ENDPOINT_CONFIG: &str = r#"{"b": 1, "a": {"z": true, "y": null}, "Z": "z", "A": "a"}"#; + const PUBLIC_PLANE: &str = "ops/dp/public/aws-us-west-2-c1"; + + // Alice owns draft ..01 (one queued discover, and one discover per stored + // status) and draft ..02 (one discover, deleted by the cascade case). Bob + // owns draft ..03 with one discover. Bob needs only an auth.users row, not + // the bob_co fixture's tenant, so he is created here as invite_links does. + // + // Grants are seeded before the server starts because its authorization + // Snapshot is taken once at start-up: bob holds legacy `write` on aliceCo/, + // carol holds the `editor` bundle with no legacy capability, and a service + // account holds `admin`. + async fn seed(pool: &sqlx::PgPool) { + sqlx::raw_sql( + r#" + INSERT INTO auth.users (id, email) VALUES + ('22222222-2222-2222-2222-222222222222', 'bob@example.test'), + ('33333333-3333-3333-3333-333333333333', 'aliceCo/robot@service_accounts.estuary.dev'), + ('44444444-4444-4444-4444-444444444444', 'carol@example.test'); + + INSERT INTO internal.service_accounts (user_id, catalog_name, created_by) VALUES + ('33333333-3333-3333-3333-333333333333', 'aliceCo/robot', '11111111-1111-1111-1111-111111111111'); + + INSERT INTO user_grants (user_id, object_role, capability, bundles) VALUES + ('22222222-2222-2222-2222-222222222222', 'aliceCo/', 'write', '{}'), + ('33333333-3333-3333-3333-333333333333', 'aliceCo/', 'admin', '{}'), + ('44444444-4444-4444-4444-444444444444', 'aliceCo/', 'none', '{editor}'); + + INSERT INTO drafts (id, user_id, detail) VALUES + ('0100000000000001', '11111111-1111-1111-1111-111111111111', 'alice draft'), + ('0100000000000002', '11111111-1111-1111-1111-111111111111', 'alice doomed draft'), + ('0100000000000003', '22222222-2222-2222-2222-222222222222', 'bob draft'); + + INSERT INTO discovers ( + id, draft_id, capture_name, connector_tag_id, data_plane_name, + endpoint_config, update_only, detail, job_status + ) + SELECT + s.id::flowid, s.draft_id::flowid, s.capture_name, '66:66:66:66:00:00:00:00', + 'ops/dp/public/aws-us-west-2-c1', '{}', s.update_only, s.detail, + ('{"type": "' || s.status || '"}')::jsonb_obj + FROM (VALUES + ('0200000000000001', '0100000000000001', 'aliceCo/in/capture-foo', true, 'first discover', 'queued'), + ('0200000000000002', '0100000000000002', 'aliceCo/in/doomed', false, null, 'queued'), + ('0200000000000003', '0100000000000003', 'bobCo/capture', false, 'bob discover', 'queued'), + + ('0300000000000001', '0100000000000001', 'aliceCo/in/s', false, null, 'queued'), + ('0300000000000002', '0100000000000001', 'aliceCo/in/s', false, null, 'success'), + ('0300000000000003', '0100000000000001', 'aliceCo/in/s', false, null, 'discoverFailed'), + ('0300000000000004', '0100000000000001', 'aliceCo/in/s', false, null, 'mergeFailed'), + ('0300000000000005', '0100000000000001', 'aliceCo/in/s', false, null, 'pullFailed'), + ('0300000000000006', '0100000000000001', 'aliceCo/in/s', false, null, 'tagFailed'), + ('0300000000000007', '0100000000000001', 'aliceCo/in/s', false, null, 'wrongProtocol'), + ('0300000000000008', '0100000000000001', 'aliceCo/in/s', false, null, 'imageForbidden'), + ('0300000000000009', '0100000000000001', 'aliceCo/in/s', false, null, 'noDataPlane'), + ('030000000000000a', '0100000000000001', 'aliceCo/in/s', false, null, 'notAuthorized'), + ('030000000000000b', '0100000000000001', 'aliceCo/in/s', false, null, 'deprecatedBackground') + ) AS s(id, draft_id, capture_name, update_only, detail, status); + "#, + ) + .execute(pool) + .await + .unwrap(); + } + + async fn query( + server: &test_server::TestServer, + id: &str, + token: Option<&str>, + ) -> serde_json::Value { + server + .graphql( + &serde_json::json!({ + "query": DISCOVER_QUERY, + "variables": { "id": id }, + }), + token, + ) + .await + } + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../../../fixtures", + scripts("data_planes", "alice", "connectors") + ) + )] + async fn test_discover_query(pool: sqlx::PgPool) { + let _guard = test_server::init(); + seed(&pool).await; + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let alice_token = server.make_access_token(ALICE, Some("alice@example.com")); + + // Alice reads her own queued discover. + let own = query(&server, "0200000000000001", Some(&alice_token)).await; + insta::assert_json_snapshot!("own_discover", own, { + ".data.discover.logsToken" => "[uuid]", + ".data.discover.createdAt" => "[datetime]", + ".data.discover.updatedAt" => "[datetime]", + }); + + // Every stored camelCase tag reads back as its GraphQL name. + let mut statuses = serde_json::Map::new(); + for id in [ + "0300000000000001", + "0300000000000002", + "0300000000000003", + "0300000000000004", + "0300000000000005", + "0300000000000006", + "0300000000000007", + "0300000000000008", + "0300000000000009", + "030000000000000a", + "030000000000000b", + ] { + let response = query(&server, id, Some(&alice_token)).await; + statuses.insert( + id.to_string(), + response["data"]["discover"]["status"].clone(), + ); + } + insta::assert_json_snapshot!("status_mapping", statuses); + + // Another user's discover and an unknown id are the same null, with no + // error to tell them apart. + let not_owned = query(&server, "0200000000000003", Some(&alice_token)).await; + let unknown = query(&server, "0f00000000000000", Some(&alice_token)).await; + assert_eq!(not_owned, unknown); + insta::assert_json_snapshot!("not_owned_or_unknown", not_owned); + + // Deleting a draft cascades to its discovers. + let before = query(&server, "0200000000000002", Some(&alice_token)).await; + assert_eq!(before["data"]["discover"]["id"], "0200000000000002"); + sqlx::query("DELETE FROM drafts WHERE id = '0100000000000002'") + .execute(&pool) + .await + .unwrap(); + let after = query(&server, "0200000000000002", Some(&alice_token)).await; + assert_eq!(after, unknown); + + // Unauthenticated callers get an error, not a null. + let unauthenticated = query(&server, "0200000000000001", None).await; + insta::assert_json_snapshot!("unauthenticated", unauthenticated); + } + + /// A `createDiscover` request. Optional arguments are omitted from the + /// document when `None`, which is the only way to exercise a default: a + /// variable bound to null is rejected for a non-null argument. The body + /// is rendered to text so that `endpoint_config` reaches the server + /// byte-for-byte; a `serde_json::Value` would sort its keys. + struct CreateRequest { + capture_name: &'static str, + connector_tag_id: &'static str, + data_plane_name: &'static str, + endpoint_config: &'static str, + draft_id: Option<&'static str>, + update_only: Option, + detail: Option<&'static str>, + } + + impl CreateRequest { + fn new(capture_name: &'static str) -> Self { + Self { + capture_name, + connector_tag_id: SOURCE_TAG, + data_plane_name: PUBLIC_PLANE, + endpoint_config: ENDPOINT_CONFIG, + draft_id: None, + update_only: None, + detail: None, + } + } + + fn body(&self) -> Box { + let quote = |s: &str| serde_json::to_string(s).unwrap(); + let mut args = format!( + "captureName: {}, connectorTagId: {}, dataPlaneName: {}, endpointConfig: $endpointConfig", + quote(self.capture_name), + quote(self.connector_tag_id), + quote(self.data_plane_name), + ); + if let Some(draft_id) = self.draft_id { + args += &format!(", draftId: {}", quote(draft_id)); + } + if let Some(update_only) = self.update_only { + args += &format!(", updateOnly: {update_only}"); + } + if let Some(detail) = self.detail { + args += &format!(", detail: {}", quote(detail)); + } + let query = format!( + "mutation($endpointConfig: JSON!) {{ createDiscover({args}) {{ \ + id draftId captureName connectorTagId dataPlaneName updateOnly detail \ + logsToken status createdAt updatedAt }} }}" + ); + serde_json::value::RawValue::from_string(format!( + r#"{{"query": {}, "variables": {{"endpointConfig": {}}}}}"#, + quote(&query), + self.endpoint_config, + )) + .unwrap() + } + } + + async fn create( + server: &test_server::TestServer, + request: CreateRequest, + token: Option<&str>, + ) -> serde_json::Value { + server.graphql(&request.body(), token).await + } + + /// The id of a successful `createDiscover`, or a panic showing the errors. + fn created_id(response: &serde_json::Value) -> String { + response["data"]["createDiscover"]["id"] + .as_str() + .unwrap_or_else(|| panic!("createDiscover failed: {response}")) + .to_string() + } + + /// What the mutation wrote, read back from the tables it touched: the + /// discovers row, its draft's owner and detail, and the automation task + /// the insert trigger enqueued. The configuration is read as text so the + /// snapshot shows field order; sqlx's JSON encoding prefixes one space + /// where JSONB would carry a version byte, which `ltrim` drops. + async fn stored(pool: &sqlx::PgPool, id: &str) -> serde_json::Value { + sqlx::query_scalar( + r#" + SELECT json_build_object( + 'capture_name', d.capture_name, + 'connector_tag_id', d.connector_tag_id::text, + 'data_plane_name', d.data_plane_name, + 'endpoint_config', ltrim(d.endpoint_config::text), + 'update_only', d.update_only, + 'detail', d.detail, + 'job_status', d.job_status, + 'draft_user_id', drafts.user_id, + 'draft_detail', drafts.detail, + 'task_type', (SELECT task_type FROM internal.tasks WHERE task_id = d.id) + ) + FROM discovers d JOIN drafts ON drafts.id = d.draft_id + WHERE d.id = $1::flowid + "#, + ) + .bind(id) + .fetch_one(pool) + .await + .unwrap() + } + + async fn count_for_capture(pool: &sqlx::PgPool, capture_name: &str) -> i64 { + sqlx::query_scalar("SELECT count(*) FROM discovers WHERE capture_name = $1") + .bind(capture_name) + .fetch_one(pool) + .await + .unwrap() + } + + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../../../fixtures", + scripts("data_planes", "alice", "connectors") + ) + )] + async fn test_create_discover(pool: sqlx::PgPool) { + let _guard = test_server::init(); + seed(&pool).await; + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let alice_token = server.make_access_token(ALICE, Some("alice@example.com")); + + // Into an existing owned draft, with every optional argument given. + // `aliceCo/in/capture-foo` has a live spec. + let request = CreateRequest { + draft_id: Some("0100000000000001"), + update_only: Some(true), + detail: Some("into existing draft"), + ..CreateRequest::new("aliceCo/in/capture-foo") + }; + let response = create(&server, request, Some(&alice_token)).await; + insta::assert_json_snapshot!("create_into_existing_draft", response, { + ".data.createDiscover.id" => "[id]", + ".data.createDiscover.logsToken" => "[uuid]", + ".data.createDiscover.createdAt" => "[datetime]", + ".data.createDiscover.updatedAt" => "[datetime]", + }); + let first_id = created_id(&response); + insta::assert_json_snapshot!( + "create_into_existing_draft_stored", + stored(&pool, &first_id).await + ); + + // Again into the same draft: no uniqueness rule applies. + let request = CreateRequest { + draft_id: Some("0100000000000001"), + ..CreateRequest::new("aliceCo/in/capture-foo") + }; + let response = create(&server, request, Some(&alice_token)).await; + let second_id = created_id(&response); + assert_ne!(first_id, second_id); + // Two created here plus the seeded one. + assert_eq!(3, count_for_capture(&pool, "aliceCo/in/capture-foo").await); + assert_eq!( + 4, + stored(&pool, &second_id).await["task_type"], + "each discover enqueues its own task" + ); + + // Draft omitted: a new draft owned by the caller is created, and + // `detail` lands on both rows. The capture has no live spec. + let request = CreateRequest { + detail: Some("auto draft"), + ..CreateRequest::new("aliceCo/brand-new/capture") + }; + let response = create(&server, request, Some(&alice_token)).await; + let created = &response["data"]["createDiscover"]; + let row = stored(&pool, &created_id(&response)).await; + let draft_id: String = sqlx::query_scalar( + "SELECT replace(id::text, ':', '') FROM drafts WHERE user_id = $1 AND detail = 'auto draft'", + ) + .bind(ALICE) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!(created["draftId"], draft_id); + insta::assert_json_snapshot!("create_auto_draft_stored", row); + + // Draft and detail both omitted: both detail columns are null. + let response = create( + &server, + CreateRequest::new("aliceCo/brand-new/capture"), + Some(&alice_token), + ) + .await; + let row = stored(&pool, &created_id(&response)).await; + assert_eq!(serde_json::Value::Null, row["detail"]); + assert_eq!(serde_json::Value::Null, row["draft_detail"]); + assert_eq!(false, row["update_only"]); + + // A service account is an ordinary caller: it owns the created draft. + let sa_token = server.make_access_token(SERVICE_ACCOUNT, None); + let response = create( + &server, + CreateRequest::new("aliceCo/robot/capture"), + Some(&sa_token), + ) + .await; + let row = stored(&pool, &created_id(&response)).await; + assert_eq!(SERVICE_ACCOUNT.to_string(), row["draft_user_id"]); + + // The `editor` bundle alone conveys SpecEdit, so carol passes the + // capture gate, but it does not reach the public plane: the tenant's + // role grant to `ops/dp/public/` is legacy `read`, whose Viewer bits + // include ViewDataPlanePrivateNetworking, and a Delegate edge passes + // through only the bits its parent holds. Editor lacks that bit, so + // the plane check fails exactly as the executor's would. + let carol_token = server.make_access_token(CAROL, Some("carol@example.test")); + let response = create( + &server, + CreateRequest::new("aliceCo/carol/capture"), + Some(&carol_token), + ) + .await; + insta::assert_json_snapshot!("create_editor_bundle_cannot_reach_plane", response); + assert_eq!(0, count_for_capture(&pool, "aliceCo/carol/capture").await); + + // No SpecEdit on the capture: alice holds nothing under bobCo/. + let denied = create(&server, CreateRequest::new("bobCo/x"), Some(&alice_token)).await; + insta::assert_json_snapshot!("create_denied_no_spec_edit", denied); + assert_eq!(0, count_for_capture(&pool, "bobCo/x").await); + + // Legacy `write` reads and appends but does not convey SpecEdit. + let bob_token = server.make_access_token(BOB, Some("bob@example.test")); + let denied = create(&server, CreateRequest::new("aliceCo/x"), Some(&bob_token)).await; + insta::assert_json_snapshot!("create_denied_legacy_write", denied); + assert_eq!(0, count_for_capture(&pool, "aliceCo/x").await); + } + + // Without a read grant on the plane, an existing plane and a missing one + // yield the same error. The grant is removed before the server starts + // because the Snapshot is taken once. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../../../fixtures", + scripts("data_planes", "alice", "connectors") + ) + )] + async fn test_create_discover_without_plane_grant(pool: sqlx::PgPool) { + let _guard = test_server::init(); + sqlx::query("DELETE FROM role_grants WHERE subject_role = 'aliceCo/' AND object_role = 'ops/dp/public/'") + .execute(&pool) + .await + .unwrap(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let alice_token = server.make_access_token(ALICE, Some("alice@example.com")); + + let ungranted = create( + &server, + CreateRequest::new("aliceCo/in/capture-foo"), + Some(&alice_token), + ) + .await; + let request = CreateRequest { + data_plane_name: "ops/dp/public/nope", + ..CreateRequest::new("aliceCo/in/capture-foo") + }; + let missing = create(&server, request, Some(&alice_token)).await; + + insta::assert_json_snapshot!( + "create_plane_not_found", + serde_json::json!({ + "ungranted": ungranted, + "missing": missing, + }) + ); + assert_eq!(0, count_for_capture(&pool, "aliceCo/in/capture-foo").await); + } + + // A denial against a Snapshot older than the request is provisional: the + // server answers 307 so the client retries once a fresh Snapshot exists, + // and nothing is written. The gated test Snapshot serves an empty + // Snapshot first, and `reqwest` must not follow the redirect for the test + // to observe it. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../../../fixtures", + scripts("data_planes", "alice", "connectors") + ) + )] + async fn test_create_discover_stale_snapshot(pool: sqlx::PgPool) { + let _guard = test_server::init(); + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let alice_token = server.make_access_token(ALICE, Some("alice@example.com")); + + let client = flow_client_next::rest::Client { + base_url: server.base_url(), + http_client: reqwest::Client::builder() + .redirect(reqwest::redirect::Policy::none()) + .build() + .unwrap(), + }; + let response = client + .post( + "/api/graphql", + &CreateRequest::new("aliceCo/in/capture-foo").body(), + Some(&alice_token), + ) + .send() + .await + .unwrap(); + + assert_eq!(reqwest::StatusCode::TEMPORARY_REDIRECT, response.status()); + let location = response.headers()["location"].to_str().unwrap(); + assert!( + location.starts_with("/api/graphql?started=") && location.contains("&retryAfter="), + "unexpected Location: {location}" + ); + assert_eq!(0, count_for_capture(&pool, "aliceCo/in/capture-foo").await); + } + + // Every synchronous rejection of `createDiscover`, pinned as one contract: + // the messages are what clients will match on, and none of them writes a + // row. Missing and not-owned drafts must be byte-identical so a draft id + // cannot probe for other users' drafts. + #[sqlx::test( + migrations = "../../supabase/migrations", + fixtures( + path = "../../../fixtures", + scripts("data_planes", "alice", "connectors") + ) + )] + async fn test_create_discover_validation_errors(pool: sqlx::PgPool) { + let _guard = test_server::init(); + seed(&pool).await; + let server = test_server::TestServer::start( + pool.clone(), + test_server::snapshot(pool.clone(), true).await, + ) + .await; + let alice_token = server.make_access_token(ALICE, Some("alice@example.com")); + let seeded: (i64, i64) = sqlx::query_as( + "SELECT (SELECT count(*) FROM discovers), (SELECT count(*) FROM drafts)", + ) + .fetch_one(&pool) + .await + .unwrap(); + + let invalid_capture_name = create( + &server, + CreateRequest::new("aliceCo/not a valid name"), + Some(&alice_token), + ) + .await; + let unknown_tag = create( + &server, + CreateRequest { + connector_tag_id: "0f0f0f0f0f0f0f0f", + ..CreateRequest::new("aliceCo/x") + }, + Some(&alice_token), + ) + .await; + // `source/multi-tag-test:v2` is a tag whose spec job failed, so its + // `protocol` is still null. + let unprocessed_tag = create( + &server, + CreateRequest { + connector_tag_id: "6666666600000005", + ..CreateRequest::new("aliceCo/x") + }, + Some(&alice_token), + ) + .await; + // `materialize/test:test` is successfully processed but the wrong protocol. + let materialization_tag = create( + &server, + CreateRequest { + connector_tag_id: "6666666600000001", + ..CreateRequest::new("aliceCo/x") + }, + Some(&alice_token), + ) + .await; + let unknown_plane = create( + &server, + CreateRequest { + data_plane_name: "ops/dp/public/nope", + ..CreateRequest::new("aliceCo/x") + }, + Some(&alice_token), + ) + .await; + // Draft ..03 belongs to bob. + let draft_not_owned = create( + &server, + CreateRequest { + draft_id: Some("0100000000000003"), + ..CreateRequest::new("aliceCo/x") + }, + Some(&alice_token), + ) + .await; + let draft_missing = create( + &server, + CreateRequest { + draft_id: Some("0100000000000f0f"), + ..CreateRequest::new("aliceCo/x") + }, + Some(&alice_token), + ) + .await; + assert_eq!(draft_not_owned, draft_missing); + let unauthenticated = create(&server, CreateRequest::new("aliceCo/x"), None).await; + + insta::assert_json_snapshot!( + "create_validation_errors", + serde_json::json!({ + "invalid_capture_name": invalid_capture_name, + "unknown_tag": unknown_tag, + "unprocessed_tag": unprocessed_tag, + "materialization_tag": materialization_tag, + "unknown_plane": unknown_plane, + "draft_not_owned_or_missing": draft_not_owned, + "unauthenticated": unauthenticated, + }) + ); + + let after: (i64, i64) = sqlx::query_as( + "SELECT (SELECT count(*) FROM discovers), (SELECT count(*) FROM drafts)", + ) + .fetch_one(&pool) + .await + .unwrap(); + assert_eq!( + seeded, after, + "a rejected discover must insert neither a discover nor a draft" + ); + } +} diff --git a/crates/control-plane-api/src/server/public/graphql/mod.rs b/crates/control-plane-api/src/server/public/graphql/mod.rs index 2828072a2c7..24402e1aa4f 100644 --- a/crates/control-plane-api/src/server/public/graphql/mod.rs +++ b/crates/control-plane-api/src/server/public/graphql/mod.rs @@ -31,6 +31,7 @@ mod data_planes; mod filters; pub(crate) use data_planes::parse_data_plane_name; mod connectors; +mod discovers; pub mod id; mod invite_links; mod live_spec_refs; @@ -123,6 +124,7 @@ pub struct QueryRoot( data_planes::DataPlanesQuery, invite_links::InviteLinksQuery, connectors::ConnectorsQuery, + discovers::DiscoversQuery, tenant::TenantQuery, refresh_tokens::RefreshTokensQuery, service_accounts::ServiceAccountsQuery, @@ -138,6 +140,7 @@ pub struct MutationRoot( alert_subscriptions::AlertSubscriptionsMutation, invite_links::InviteLinksMutation, data_planes::DataPlanesMutation, + discovers::DiscoversMutation, refresh_tokens::RefreshTokensMutation, service_accounts::ServiceAccountsMutation, secrets::SecretsMutation, diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_auto_draft_stored.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_auto_draft_stored.snap new file mode 100644 index 00000000000..61199146463 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_auto_draft_stored.snap @@ -0,0 +1,18 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: row +--- +{ + "capture_name": "aliceCo/brand-new/capture", + "connector_tag_id": "66:66:66:66:00:00:00:00", + "data_plane_name": "ops/dp/public/aws-us-west-2-c1", + "detail": "auto draft", + "draft_detail": "auto draft", + "draft_user_id": "11111111-1111-1111-1111-111111111111", + "endpoint_config": "{\"b\":1,\"a\":{\"z\":true,\"y\":null},\"Z\":\"z\",\"A\":\"a\"}", + "job_status": { + "type": "queued" + }, + "task_type": 4, + "update_only": false +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_denied_legacy_write.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_denied_legacy_write.snap new file mode 100644 index 00000000000..85497c93a25 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_denied_legacy_write.snap @@ -0,0 +1,21 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: denied +--- +{ + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "PermissionDenied: bob@example.test is not authorized to access prefix or name 'aliceCo/x' with required capability SpecEdit", + "path": [ + "createDiscover" + ] + } + ] +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_denied_no_spec_edit.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_denied_no_spec_edit.snap new file mode 100644 index 00000000000..46825587ba6 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_denied_no_spec_edit.snap @@ -0,0 +1,21 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: denied +--- +{ + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "PermissionDenied: alice@example.com is not authorized to access prefix or name 'bobCo/x' with required capability SpecEdit", + "path": [ + "createDiscover" + ] + } + ] +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_editor_bundle_cannot_reach_plane.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_editor_bundle_cannot_reach_plane.snap new file mode 100644 index 00000000000..ca0624db860 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_editor_bundle_cannot_reach_plane.snap @@ -0,0 +1,21 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: response +--- +{ + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "data plane ops/dp/public/aws-us-west-2-c1 was not found", + "path": [ + "createDiscover" + ] + } + ] +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_into_existing_draft.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_into_existing_draft.snap new file mode 100644 index 00000000000..d19934a7c95 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_into_existing_draft.snap @@ -0,0 +1,21 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: response +--- +{ + "data": { + "createDiscover": { + "captureName": "aliceCo/in/capture-foo", + "connectorTagId": "6666666600000000", + "createdAt": "[datetime]", + "dataPlaneName": "ops/dp/public/aws-us-west-2-c1", + "detail": "into existing draft", + "draftId": "0100000000000001", + "id": "[id]", + "logsToken": "[uuid]", + "status": "QUEUED", + "updateOnly": true, + "updatedAt": "[datetime]" + } + } +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_into_existing_draft_stored.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_into_existing_draft_stored.snap new file mode 100644 index 00000000000..05b048e6ef0 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_into_existing_draft_stored.snap @@ -0,0 +1,18 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: "stored(&pool, &first_id).await" +--- +{ + "capture_name": "aliceCo/in/capture-foo", + "connector_tag_id": "66:66:66:66:00:00:00:00", + "data_plane_name": "ops/dp/public/aws-us-west-2-c1", + "detail": "into existing draft", + "draft_detail": "alice draft", + "draft_user_id": "11111111-1111-1111-1111-111111111111", + "endpoint_config": "{\"b\":1,\"a\":{\"z\":true,\"y\":null},\"Z\":\"z\",\"A\":\"a\"}", + "job_status": { + "type": "queued" + }, + "task_type": 4, + "update_only": true +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_plane_not_found.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_plane_not_found.snap new file mode 100644 index 00000000000..d5179423408 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_plane_not_found.snap @@ -0,0 +1,40 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: "serde_json::json!({ \"ungranted\": ungranted, \"missing\": missing, })" +--- +{ + "missing": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "data plane ops/dp/public/nope was not found", + "path": [ + "createDiscover" + ] + } + ] + }, + "ungranted": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "data plane ops/dp/public/aws-us-west-2-c1 was not found", + "path": [ + "createDiscover" + ] + } + ] + } +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_validation_errors.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_validation_errors.snap new file mode 100644 index 00000000000..eb31815d266 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__create_validation_errors.snap @@ -0,0 +1,125 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: "serde_json::json!({\n \"invalid_capture_name\": invalid_capture_name, \"unknown_tag\": unknown_tag,\n \"unprocessed_tag\": unprocessed_tag, \"materialization_tag\":\n materialization_tag, \"unknown_plane\": unknown_plane,\n \"draft_not_owned_or_missing\": draft_not_owned, \"unauthenticated\":\n unauthenticated,\n})" +--- +{ + "draft_not_owned_or_missing": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "draft not found", + "path": [ + "createDiscover" + ] + } + ] + }, + "invalid_capture_name": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "invalid capture name: : aliceCo/not a valid name doesn't match pattern [\\p{Letter}\\p{Number}\\-_\\.]+(/[\\p{Letter}\\p{Number}\\-_\\.]+)* (unmatched portion is: a valid name)", + "path": [ + "createDiscover" + ] + } + ] + }, + "materialization_tag": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "connector tag 6666666600000001 is not a capture connector", + "path": [ + "createDiscover" + ] + } + ] + }, + "unauthenticated": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "status: 'The request does not have valid authentication credentials', self: \"This is an authenticated API but the request is missing a required Authorization: Bearer token\"", + "path": [ + "createDiscover" + ] + } + ] + }, + "unknown_plane": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "data plane ops/dp/public/nope was not found", + "path": [ + "createDiscover" + ] + } + ] + }, + "unknown_tag": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "connector tag not found", + "path": [ + "createDiscover" + ] + } + ] + }, + "unprocessed_tag": { + "data": null, + "errors": [ + { + "locations": [ + { + "column": 36, + "line": 1 + } + ], + "message": "connector tag 6666666600000005 has not been successfully processed", + "path": [ + "createDiscover" + ] + } + ] + } +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__not_owned_or_unknown.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__not_owned_or_unknown.snap new file mode 100644 index 00000000000..41efa6b19fc --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__not_owned_or_unknown.snap @@ -0,0 +1,9 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: not_owned +--- +{ + "data": { + "discover": null + } +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__own_discover.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__own_discover.snap new file mode 100644 index 00000000000..b10d0cf2655 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__own_discover.snap @@ -0,0 +1,21 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: own +--- +{ + "data": { + "discover": { + "captureName": "aliceCo/in/capture-foo", + "connectorTagId": "6666666600000000", + "createdAt": "[datetime]", + "dataPlaneName": "ops/dp/public/aws-us-west-2-c1", + "detail": "first discover", + "draftId": "0100000000000001", + "id": "0200000000000001", + "logsToken": "[uuid]", + "status": "QUEUED", + "updateOnly": true, + "updatedAt": "[datetime]" + } + } +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__status_mapping.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__status_mapping.snap new file mode 100644 index 00000000000..83e42ad558e --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__status_mapping.snap @@ -0,0 +1,17 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: statuses +--- +{ + "0300000000000001": "QUEUED", + "0300000000000002": "SUCCESS", + "0300000000000003": "DISCOVER_FAILED", + "0300000000000004": "MERGE_FAILED", + "0300000000000005": "PULL_FAILED", + "0300000000000006": "TAG_FAILED", + "0300000000000007": "WRONG_PROTOCOL", + "0300000000000008": "IMAGE_FORBIDDEN", + "0300000000000009": "NO_DATA_PLANE", + "030000000000000a": "NOT_AUTHORIZED", + "030000000000000b": "DEPRECATED_BACKGROUND" +} diff --git a/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__unauthenticated.snap b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__unauthenticated.snap new file mode 100644 index 00000000000..0e4a0403bb0 --- /dev/null +++ b/crates/control-plane-api/src/server/public/graphql/snapshots/control_plane_api__server__public__graphql__discovers__test__unauthenticated.snap @@ -0,0 +1,21 @@ +--- +source: crates/control-plane-api/src/server/public/graphql/discovers.rs +expression: unauthenticated +--- +{ + "data": null, + "errors": [ + { + "locations": [ + { + "column": 9, + "line": 3 + } + ], + "message": "status: 'The request does not have valid authentication credentials', self: \"This is an authenticated API but the request is missing a required Authorization: Bearer token\"", + "path": [ + "discover" + ] + } + ] +} diff --git a/crates/flow-client/control-plane-api.graphql b/crates/flow-client/control-plane-api.graphql index 4253c0d7a47..a7a37139d28 100644 --- a/crates/flow-client/control-plane-api.graphql +++ b/crates/flow-client/control-plane-api.graphql @@ -411,6 +411,8 @@ enum CapabilityBit { Assume } +scalar Capture + type CardPaymentMethodDetails { brand: String last4: String @@ -866,6 +868,36 @@ The input/output is a string in RFC3339 format. """ scalar DateTime +""" +A user-initiated connector discovery operation. +""" +type Discover { + id: Id! + """ + The draft that discovered specs are merged into. + """ + draftId: Id! + """ + Catalog name of the capture being discovered. + """ + captureName: Capture! + connectorTagId: Id! + dataPlaneName: Name! + updateOnly: Boolean! + detail: String + """ + Bearer token for reading this operation's logs. + """ + logsToken: UUID! + """ + Current outcome of the operation. `QUEUED` until the server has + processed it. + """ + status: DiscoverStatusType! + createdAt: DateTime! + updatedAt: DateTime! +} + """ A capture binding that has changed as a result of a discover """ @@ -884,6 +916,63 @@ type DiscoverChange { disable: Boolean! } +""" +Current outcome of a discover operation. + +This is the unit-variant projection of [`JobStatus`], which +`async_graphql::Enum` requires and which `JobStatus::Success` (carrying +data) cannot satisfy. Its serde names are the same camelCase tags stored in +`discovers.job_status`, so a stored `type` parses into it directly; only the +GraphQL representation is `SCREAMING_SNAKE_CASE`. Variants are ordered for +the published schema rather than in `JobStatus` order. +""" +enum DiscoverStatusType { + """ + The discover has not yet been processed. + """ + QUEUED + """ + Discovered bindings were merged into the draft. + """ + SUCCESS + """ + The connector's discovery RPC failed. + """ + DISCOVER_FAILED + """ + Discovered specs could not be merged with the draft's existing specs. + """ + MERGE_FAILED + """ + The connector image could not be pulled. + """ + PULL_FAILED + """ + The connector tag has not been successfully processed. + """ + TAG_FAILED + """ + The connector is not a capture connector. + """ + WRONG_PROTOCOL + """ + The connector image is not allowed. + """ + IMAGE_FORBIDDEN + """ + The data plane does not exist or the caller cannot use it. + """ + NO_DATA_PLANE + """ + The caller does not hold `SpecEdit` on the capture name. + """ + NOT_AUTHORIZED + """ + The discover used the deprecated background flag. + """ + DEPRECATED_BACKGROUND +} + type EffectiveAlertConfig { config: JSON! provenance: [FieldProvenance!]! @@ -1380,6 +1469,56 @@ type MutationRoot { """ removeDataPlanePrivateLink(id: Id!): Id! """ + Queue a connector discovery for `captureName`, merging the discovered + bindings and collections into a draft. + + When `draftId` is given, that draft is used and must be owned by the + caller. When it is omitted, a new draft owned by the caller is created + and its id is returned as `Discover.draftId`. + + The caller must hold `SpecEdit` on `captureName` and read on + `dataPlaneName`. The connector tag must exist, have a successfully + processed spec, and be a capture connector. Poll `discover(id)` for the + outcome; the merged specs are in the draft. + + Note that a legacy `write` grant does not convey `SpecEdit`: users who + can read and append to a prefix's collections still cannot discover + under it. Legacy `admin` on a prefix covering the capture satisfies + both requirements in a normally provisioned tenant. The `Editor` bundle + conveys `SpecEdit` but on its own does not reach the public data + planes through the tenant's role grant. + """ + createDiscover( + """ + Catalog name of the capture to discover. It need not exist yet. + """ + captureName: Capture!, + """ + Id of the connector tag to run, as returned by `connector.spec(imageTag).id`. + """ + connectorTagId: Id!, + """ + Data plane in which to run the connector. The caller must hold read on it. + """ + dataPlaneName: Name!, + """ + Endpoint configuration passed to the connector, stored verbatim. Object field order is preserved so that sops-encrypted configurations remain decryptable. Encryption is the caller's responsibility. + """ + endpointConfig: JSON!, + """ + The draft that discovered specs are merged into. Must be owned by the caller. When omitted, a new draft is created for this discover. + """ + draftId: Id, + """ + When true, existing bindings are refreshed and newly discovered bindings are added in a disabled state. + """ + updateOnly: Boolean! = false, + """ + Optional description, recorded on the discover and on a draft this mutation creates. + """ + detail: String + ): Discover! + """ Create a refresh token for the authenticated user. Service-account callers are rejected: their API keys are administered @@ -1977,6 +2116,12 @@ type QueryRoot { Returns a paginated list of connectors, optionally filtered by protocol. """ connectors(filter: ConnectorsFilter, after: String, before: String, first: Int, last: Int): ConnectorConnection! + """ + Fetch a single discover by id. Returns null if no such discover exists + or the caller does not own its draft; the two cases are + indistinguishable. + """ + discover(id: Id!): Discover tenant(name: String!): Tenant """ List refresh tokens owned by the authenticated user. diff --git a/crates/models/src/discovers.rs b/crates/models/src/discovers.rs index 9901376ba5d..c4bac7858a7 100644 --- a/crates/models/src/discovers.rs +++ b/crates/models/src/discovers.rs @@ -1,6 +1,6 @@ use std::collections::BTreeMap; -use crate::ResourcePath; +use crate::{Id, ResourcePath}; /// Represents a capture binding that was added, removed, or modified by a /// discover. @@ -15,3 +15,144 @@ pub struct Changed { } /// Represents a set of changes resulting from a discover. pub type Changes = BTreeMap; + +/// JobStatus is the possible outcomes of a handled discover operation. +/// +/// It is stored verbatim in `discovers.job_status`, which PostgREST clients +/// read directly, so the camelCase `type` tags are a wire contract. +#[derive(Debug, serde::Deserialize, serde::Serialize)] +#[serde(rename_all = "camelCase", tag = "type")] +pub enum JobStatus { + Queued, + WrongProtocol, + TagFailed, + ImageForbidden, + PullFailed, + DiscoverFailed, + MergeFailed, + Success { + #[serde(default, skip_serializing_if = "Option::is_none")] + publication_id: Option, + #[serde(default, skip_serializing_if = "std::ops::Not::not")] + specs_unchanged: bool, + }, + DeprecatedBackground, + NoDataPlane, + NotAuthorized, +} + +impl JobStatus { + pub fn is_success(&self) -> bool { + matches!(self, JobStatus::Success { .. }) + } +} + +/// Current outcome of a discover operation. +/// +/// This is the unit-variant projection of [`JobStatus`], which +/// `async_graphql::Enum` requires and which `JobStatus::Success` (carrying +/// data) cannot satisfy. Its serde names are the same camelCase tags stored in +/// `discovers.job_status`, so a stored `type` parses into it directly; only the +/// GraphQL representation is `SCREAMING_SNAKE_CASE`. Variants are ordered for +/// the published schema rather than in `JobStatus` order. +#[derive(Debug, Clone, Copy, PartialEq, Eq, serde::Deserialize, serde::Serialize)] +#[cfg_attr( + feature = "async-graphql", + derive(async_graphql::Enum), + graphql(rename_items = "SCREAMING_SNAKE_CASE") +)] +#[serde(rename_all = "camelCase")] +pub enum DiscoverStatusType { + /// The discover has not yet been processed. + Queued, + /// Discovered bindings were merged into the draft. + Success, + /// The connector's discovery RPC failed. + DiscoverFailed, + /// Discovered specs could not be merged with the draft's existing specs. + MergeFailed, + /// The connector image could not be pulled. + PullFailed, + /// The connector tag has not been successfully processed. + TagFailed, + /// The connector is not a capture connector. + WrongProtocol, + /// The connector image is not allowed. + ImageForbidden, + /// The data plane does not exist or the caller cannot use it. + NoDataPlane, + /// The caller does not hold `SpecEdit` on the capture name. + NotAuthorized, + /// The discover used the deprecated background flag. + DeprecatedBackground, +} + +impl From<&JobStatus> for DiscoverStatusType { + fn from(status: &JobStatus) -> Self { + match status { + JobStatus::Queued => DiscoverStatusType::Queued, + JobStatus::Success { .. } => DiscoverStatusType::Success, + JobStatus::DiscoverFailed => DiscoverStatusType::DiscoverFailed, + JobStatus::MergeFailed => DiscoverStatusType::MergeFailed, + JobStatus::PullFailed => DiscoverStatusType::PullFailed, + JobStatus::TagFailed => DiscoverStatusType::TagFailed, + JobStatus::WrongProtocol => DiscoverStatusType::WrongProtocol, + JobStatus::ImageForbidden => DiscoverStatusType::ImageForbidden, + JobStatus::NoDataPlane => DiscoverStatusType::NoDataPlane, + JobStatus::NotAuthorized => DiscoverStatusType::NotAuthorized, + JobStatus::DeprecatedBackground => DiscoverStatusType::DeprecatedBackground, + } + } +} + +#[cfg(test)] +mod test { + use super::{DiscoverStatusType, JobStatus}; + + // `discovers.job_status` is a UI-facing contract: pin every wire tag, and + // pin that each tag parses into the DiscoverStatusType that + // `From<&JobStatus>` yields, since the two are meant to be interchangeable + // views of the stored `type`. The exhaustive `From` match is the + // compile-time guard for a new JobStatus variant; this list is the wire + // guard and must be extended alongside it. + #[test] + fn test_job_status_wire_tags() { + let all = vec![ + JobStatus::Queued, + JobStatus::WrongProtocol, + JobStatus::TagFailed, + JobStatus::ImageForbidden, + JobStatus::PullFailed, + JobStatus::DiscoverFailed, + JobStatus::MergeFailed, + JobStatus::Success { + publication_id: None, + specs_unchanged: false, + }, + JobStatus::Success { + publication_id: Some(crate::Id::new([1, 2, 3, 4, 5, 6, 7, 8])), + specs_unchanged: true, + }, + JobStatus::DeprecatedBackground, + JobStatus::NoDataPlane, + JobStatus::NotAuthorized, + ]; + let serialized = serde_json::to_value(&all).unwrap(); + insta::assert_json_snapshot!("job-status-wire-tags", serialized); + + // Every tag round-trips to a variant that serializes identically. + let round: Vec = serde_json::from_value(serialized.clone()).unwrap(); + assert_eq!(serialized, serde_json::to_value(&round).unwrap()); + + let mapping: Vec<(String, DiscoverStatusType)> = all + .iter() + .map(|status| { + let tag = serde_json::to_value(status).unwrap()["type"].clone(); + let parsed: DiscoverStatusType = serde_json::from_value(tag.clone()).unwrap(); + assert_eq!(DiscoverStatusType::from(status), parsed, "tag {tag}"); + (tag.as_str().unwrap().to_string(), parsed) + }) + .collect(); + insta::assert_debug_snapshot!("job-status-to-discover-status-type", mapping); + } +} diff --git a/crates/models/src/snapshots/models__discovers__test__job-status-to-discover-status-type.snap b/crates/models/src/snapshots/models__discovers__test__job-status-to-discover-status-type.snap new file mode 100644 index 00000000000..3b30872c104 --- /dev/null +++ b/crates/models/src/snapshots/models__discovers__test__job-status-to-discover-status-type.snap @@ -0,0 +1,54 @@ +--- +source: crates/models/src/discovers.rs +expression: mapping +--- +[ + ( + "queued", + Queued, + ), + ( + "wrongProtocol", + WrongProtocol, + ), + ( + "tagFailed", + TagFailed, + ), + ( + "imageForbidden", + ImageForbidden, + ), + ( + "pullFailed", + PullFailed, + ), + ( + "discoverFailed", + DiscoverFailed, + ), + ( + "mergeFailed", + MergeFailed, + ), + ( + "success", + Success, + ), + ( + "success", + Success, + ), + ( + "deprecatedBackground", + DeprecatedBackground, + ), + ( + "noDataPlane", + NoDataPlane, + ), + ( + "notAuthorized", + NotAuthorized, + ), +] diff --git a/crates/models/src/snapshots/models__discovers__test__job-status-wire-tags.snap b/crates/models/src/snapshots/models__discovers__test__job-status-wire-tags.snap new file mode 100644 index 00000000000..b9f6f4e51c3 --- /dev/null +++ b/crates/models/src/snapshots/models__discovers__test__job-status-wire-tags.snap @@ -0,0 +1,44 @@ +--- +source: crates/models/src/discovers.rs +expression: serialized +--- +[ + { + "type": "queued" + }, + { + "type": "wrongProtocol" + }, + { + "type": "tagFailed" + }, + { + "type": "imageForbidden" + }, + { + "type": "pullFailed" + }, + { + "type": "discoverFailed" + }, + { + "type": "mergeFailed" + }, + { + "type": "success" + }, + { + "publication_id": "0102030405060708", + "specs_unchanged": true, + "type": "success" + }, + { + "type": "deprecatedBackground" + }, + { + "type": "noDataPlane" + }, + { + "type": "notAuthorized" + } +]