flowctl: update raw discover/capture to use runtime crate

Also expand the perview of `flowctl generate` to also generate endpoint
config and resource config stubs for captures and materializations where
the indicate files are missing.

This is a replacement for the two-step "discover" workflow we've
previously had, where the first stage would write a config and the
second would actually do discover. That doesn't work with local
connectors (because we need a proper capture spec in order to know what
connector to discover from).

crates/flowctl/src/lib.rs

@@ -8,7 +8,6 @@ mod auth;
 mod catalog;
 mod collection;
 mod config;
-mod connector;
 mod controlplane;
 mod dataplane;
 mod draft;
@@ -272,3 +271,7 @@ fn format_user(email: Option<String>, full_name: Option<String>, id: Option<uuid
         id = id.map(|id| id.to_string()).unwrap_or_default(),
     )
 }
+
+fn status_to_anyhow(status: tonic::Status) -> anyhow::Error {
+    anyhow::anyhow!(status.message().to_string())
+}

crates/flowctl/src/preview/mod.rs

@@ -11,7 +11,7 @@ use tokio::sync::broadcast;
 #[derive(Debug, clap::Args)]
 #[clap(rename_all = "kebab-case")]
 pub struct Preview {
-    /// Path or URL to a Flow specification file to generate development files for.
+    /// Path or URL to a Flow specification file.
     #[clap(long)]
     source: String,
     /// Name of the derived collection to preview within the Flow specification file.
@@ -60,7 +60,7 @@ impl Preview {
         } else if sources.collections.is_empty() {
             anyhow::bail!("sourced specification files do not contain any derivations");
         } else {
-            anyhow::bail!("sourced specification files contain multiple derivations. Use --collection to identify the specific one to preview");
+            anyhow::bail!("sourced specification files contain multiple derivations. Use --collection to identify a specific one");
         };
 
         // Resolve the built collection and its contained derivation.
@@ -158,13 +158,13 @@ impl Preview {
         )
         .serve_derive(request_rx)
         .await
-        .map_err(|status| anyhow::anyhow!("{}", status.message()))?;
+        .map_err(crate::status_to_anyhow)?;
 
         let _opened = responses_rx
             .next()
             .await
             .context("expected Opened, not EOF")?
-            .map_err(status_to_anyhow)?
+            .map_err(crate::status_to_anyhow)?
             .opened
             .context("expected Opened")?;
 
@@ -349,7 +349,7 @@ where
     let mut inferred_shape = doc::Shape::nothing();
 
     while let Some(response) = responses_rx.next().await {
-        let response = response.map_err(status_to_anyhow)?;
+        let response = response.map_err(crate::status_to_anyhow)?;
 
         let internal = response
             .get_internal()
@@ -389,7 +389,3 @@ where
         None
     })
 }
-
-fn status_to_anyhow(status: tonic::Status) -> anyhow::Error {
-    anyhow::anyhow!(status.message().to_string())
-}