Aaron/subgraph watch introspection (#2142)

<!--
First, 🌠 thank you 🌠 for taking the time to consider a contribution to
Apollo!

Here are some important details to follow:

* ⏰ Your time is important
To save your precious time, if the contribution you are making will
take more than an hour, please make sure it has been discussed in an
        issue first. This is especially true for feature requests!

* 💡 Features
Feature requests can be created and discussed within a GitHub Issue.
Be sure to search for existing feature requests (and related issues!)
prior to opening a new request. If an existing issue covers the need,
please upvote that issue by using the 👍 emote, rather than opening a
        new issue.

* 🕷 Bug fixes
These can be created and discussed in this repository. When fixing a
bug,
please _try_ to add a test which verifies the fix. If you cannot, you
should
still submit the PR but we may still ask you (and help you!) to create a
test.

* 📖 Contribution guidelines
Follow https://github.com/apollographql/rover/blob/HEAD/CONTRIBUTING.md
when submitting a pull request. Make sure existing tests still pass, and
add
        tests for all new behavior.

* ✏️ Explain your pull request
Describe the big picture of your changes here to communicate to what
        your pull request is meant to accomplish. Provide 🔗 links 🔗 to
        associated issues!

We hope you will find this to be a positive experience! Open source
contribution can be intimidating and we hope to alleviate that pain as
much
as possible. Without following these guidelines, you may be missing
context
that can help you succeed with your contribution, which is why we
encourage
discussion first. Ultimately, there is no guarantee that we will be able
to
merge your pull-request, but by following these guidelines we can try to
avoid disappointment.

-->

---------

Co-authored-by: Aaron Arinder <[email protected]>

Author: 2 people

src/command/dev/runner.rs

@@ -1,5 +1,5 @@
 use anyhow::anyhow;
-use apollo_federation_types::config::{SchemaSource, SupergraphConfig};
+use apollo_federation_types::config::SupergraphConfig;
 use futures::stream::StreamExt;
 use tokio::task::JoinHandle;
 
@@ -18,6 +18,7 @@ use crate::{
     RoverError, RoverResult,
 };
 
+// TODO: handle retry flag for subgraphs (see rover dev help)
 pub struct Runner {
     client_config: StudioClientConfig,
     supergraph_opts: SupergraphOpts,
@@ -69,27 +70,24 @@ impl Runner {
 
         // Create subgraph config watchers.
         for (subgraph, subgraph_config) in supergraph_config.into_iter() {
-            match subgraph_config.schema {
-                SchemaSource::File { ref file } => {
-                    // Create a new file watcher kind.
-                    let kind = SubgraphConfigWatcherKind::File(FileWatcher::new(file.clone()));
-                    // Construct a subgraph config watcher from the file watcher kind.
-                    let watcher = SubgraphConfigWatcher::new(kind, subgraph_config);
-                    // Create and run the file watcher in a sub task.
-                    let (mut stream, subtask) = Subtask::new(watcher);
-                    subtask.run();
-
-                    let task = tokio::task::spawn(async move {
-                        while let Some(_) = stream.next().await {
-                            eprintln!("subgraph update: {subgraph}");
-                        }
-                    });
-                    futs.push(task);
+            // FIXME: remove unwrap
+            // Create a new file watcher kind.
+            let watcher_kind: SubgraphConfigWatcherKind =
+                subgraph_config.schema.try_into().unwrap();
+
+            // Construct a subgraph config watcher from the file watcher kind.
+            let watcher = SubgraphConfigWatcher::new(watcher_kind, &subgraph);
+            // Create and run the file watcher in a sub task.
+            let (mut stream, subtask) = Subtask::new(watcher);
+            subtask.run();
+
+            let task = tokio::task::spawn(async move {
+                while let Some(_) = stream.next().await {
+                    eprintln!("subgraph update: {subgraph}");
                 }
-                SchemaSource::SubgraphIntrospection { .. } => todo!(),
-                SchemaSource::Sdl { .. } => todo!(),
-                SchemaSource::Subgraph { .. } => todo!(),
-            };
+            });
+
+            futs.push(task);
         }
 
         futs
@@ -105,7 +103,7 @@ impl Runner {
             false,
         )
         .await
-        .map_err(|_| RoverError::new(anyhow!("TODO: get actual error")))?
+        .map_err(|err| RoverError::new(anyhow!("{err}")))?
         .ok_or_else(|| RoverError::new(anyhow!("Why is supergraph config None?")))
     }
 }

src/command/dev/watcher/subgraph_config.rs

@@ -1,12 +1,19 @@
+use std::fs::OpenOptions;
 use std::{marker::Send, pin::Pin};
 
-use apollo_federation_types::config::SubgraphConfig;
+use anyhow::{anyhow, Error};
+use apollo_federation_types::config::{SchemaSource, SubgraphConfig};
+use camino::Utf8PathBuf;
+use futures::stream::BoxStream;
 use futures::{Stream, StreamExt};
 use rover_std::errln;
 use tap::TapFallible;
 use tokio::{sync::mpsc::UnboundedSender, task::AbortHandle};
 
-use crate::command::dev::{introspect::IntrospectRunnerKind, subtask::SubtaskHandleUnit};
+use crate::cli::RoverOutputFormatKind;
+use crate::command::dev::{subtask::SubtaskHandleUnit, types::SubgraphUrl};
+use crate::command::subgraph::introspect::Introspect as SubgraphIntrospect;
+use crate::options::{IntrospectOpts, OutputOpts};
 
 use super::file::FileWatcher;
 
@@ -15,31 +22,117 @@ pub enum SubgraphConfigWatcherKind {
     /// Watch a file on disk.
     File(FileWatcher),
     /// Poll an endpoint via introspection.
-    _Introspect(IntrospectRunnerKind, u64),
+    Introspect(SubgraphIntrospection),
     /// Don't ever update, schema is only pulled once.
     _Once(String),
 }
 
+#[derive(Debug, Clone)]
+pub struct SubgraphIntrospection {
+    endpoint: SubgraphUrl,
+    // TODO: ticket using a hashmap, not a tuple, in introspect opts as eventual cleanup
+    headers: Option<Vec<(String, String)>>,
+}
+
+//TODO: impl retry (needed at least for dev)
+impl SubgraphIntrospection {
+    fn new(endpoint: SubgraphUrl, headers: Option<Vec<(String, String)>>) -> Self {
+        Self { endpoint, headers }
+    }
+
+    async fn watch(&self, subgraph_name: &str) -> FileWatcher {
+        let client = reqwest::Client::new();
+
+        //FIXME: unwrap removed
+        // TODO: does this re-use tmp dirs? or, what? don't want errors second time we run
+        // TODO: clean up after?
+        let tmp_dir = tempfile::Builder::new().tempdir().unwrap();
+        let tmp_config_dir_path = Utf8PathBuf::try_from(tmp_dir.into_path()).unwrap();
+
+        // NOTE: this assumes subgraph names are unique; are they?
+        let tmp_introspection_file = tmp_config_dir_path.join(subgraph_name);
+
+        let _ = OpenOptions::new()
+            .write(true)
+            .create(true)
+            .truncate(true)
+            .open(tmp_introspection_file.clone())
+            // FIXME: unwrap
+            .unwrap();
+
+        let output_opts = OutputOpts {
+            format_kind: RoverOutputFormatKind::default(),
+            output_file: Some(tmp_introspection_file.clone()),
+        };
+
+        let endpoint = self.endpoint.clone();
+        let headers = self.headers.clone();
+
+        tokio::spawn(async move {
+            let _ = SubgraphIntrospect {
+                opts: IntrospectOpts {
+                    endpoint,
+                    headers,
+                    // TODO impl retries (at least for dev from cli flag)
+                    watch: true,
+                },
+            }
+            .run(client, &output_opts, None)
+            .await
+            .map_err(|err| anyhow!(err));
+        });
+
+        FileWatcher::new(tmp_introspection_file)
+    }
+}
+
 impl SubgraphConfigWatcherKind {
-    async fn watch(&self) -> Pin<Box<dyn Stream<Item = String> + Send>> {
+    async fn watch(&self, subgraph_name: &str) -> Pin<Box<dyn Stream<Item = String> + Send>> {
         match self {
             Self::File(file_watcher) => file_watcher.clone().watch(),
-            Self::_Introspect(_, _) => todo!(),
+            Self::Introspect(introspection) => {
+                let watcher = introspection.watch(subgraph_name).await;
+                println!("watcher: {watcher:?}");
+
+                watcher.watch()
+            }
             Self::_Once(_) => todo!(),
         }
     }
 }
 
+impl TryFrom<SchemaSource> for SubgraphConfigWatcherKind {
+    // FIXME: anyhow error -> bespoke error with impl From to rovererror or whatever
+    type Error = anyhow::Error;
+    fn try_from(schema_source: SchemaSource) -> Result<Self, Self::Error> {
+        match schema_source {
+            SchemaSource::File { file } => Ok(Self::File(FileWatcher::new(file))),
+            SchemaSource::SubgraphIntrospection {
+                subgraph_url,
+                introspection_headers,
+            } => Ok(Self::Introspect(SubgraphIntrospection {
+                endpoint: subgraph_url,
+                headers: introspection_headers.map(|header_map| header_map.into_iter().collect()),
+            })),
+            // SDL (stdin? not sure) / Subgraph (ie, from graph-ref)
+            unsupported_source => Err(anyhow!(
+                "unsupported subgraph introspection source: {unsupported_source:?}"
+            )),
+        }
+    }
+}
+
 pub struct SubgraphConfigWatcher {
+    subgraph_name: String,
     watcher: SubgraphConfigWatcherKind,
-    //subgraph_config: SubgraphConfig,
 }
 
 impl SubgraphConfigWatcher {
-    pub fn new(watcher: SubgraphConfigWatcherKind, _subgraph_config: SubgraphConfig) -> Self {
+    // not sure we need the subgraph config?
+    pub fn new(watcher: SubgraphConfigWatcherKind, subgraph_name: &str) -> Self {
         Self {
             watcher,
-            //subgraph_config,
+            subgraph_name: subgraph_name.to_string(),
         }
     }
 }
@@ -52,23 +145,27 @@ impl SubtaskHandleUnit for SubgraphConfigWatcher {
 
     fn handle(self, sender: UnboundedSender<Self::Output>) -> AbortHandle {
         tokio::spawn(async move {
-            while let Some(content) = self.watcher.watch().await.next().await {
-                let parsed_config: Result<SubgraphConfig, serde_yaml::Error> =
-                    serde_yaml::from_str(&content);
+            while let Some(content) = self.watcher.watch(&self.subgraph_name).await.next().await {
+                // TODO: fix parsing; see wtf is up
+                //let parsed_config: Result<SubgraphConfig, serde_yaml::Error> =
+                //    serde_yaml::from_str(&content);
+                let _ = sender
+                    .send(SubgraphChanged)
+                    .tap_err(|err| tracing::error!("{:?}", err));
 
                 // We're only looking at whether a subgraph has changed, but we won't emit events
                 // if the subgraph config can't be parsed to fail early for composition
-                match parsed_config {
-                    Ok(_subgraph_config) => {
-                        let _ = sender
-                            .send(SubgraphChanged)
-                            .tap_err(|err| tracing::error!("{:?}", err));
-                    }
-                    Err(err) => {
-                        tracing::error!("Could not parse subgraph config file: {:?}", err);
-                        errln!("could not parse subgraph config file");
-                    }
-                }
+                //match parsed_config {
+                //    Ok(_subgraph_config) => {
+                //        let _ = sender
+                //            .send(SubgraphChanged)
+                //            .tap_err(|err| tracing::error!("{:?}", err));
+                //    }
+                //    Err(err) => {
+                //        tracing::error!("Could not parse subgraph config file: {:?}", err);
+                //        errln!("could not parse subgraph config file");
+                //    }
+                //}
             }
         })
         .abort_handle()

src/command/subgraph/introspect.rs

@@ -56,7 +56,7 @@ impl Introspect {
         )
     }
 
-    pub async fn exec_and_watch(
+    async fn exec_and_watch(
         &self,
         client: &Client,
         output_opts: &OutputOpts,

src/command/subgraph/mod.rs

@@ -1,7 +1,7 @@
 mod check;
 mod delete;
 mod fetch;
-mod introspect;
+pub mod introspect;
 mod lint;
 mod list;
 mod publish;

src/options/output.rs

@@ -81,7 +81,7 @@ impl RoverPrinter for RoverError {
 pub struct OutputOpts {
     /// Specify Rover's format type
     #[arg(long = "format", global = true, default_value_t)]
-    format_kind: RoverOutputFormatKind,
+    pub format_kind: RoverOutputFormatKind,
 
     /// Specify a file to write Rover's output to
     #[arg(long = "output", short = 'o', global = true, value_parser = Self::parse_absolute_path)]