ROVER-245 Plumbing in subgraph change events

The LSP has to track subgraphs being added and removed
so we need to expand the language of composition events to do this.

src/command/lsp/mod.rs

@@ -5,7 +5,9 @@ use crate::composition::supergraph::config::{
     ResolveSupergraphConfigError, SupergraphConfigResolver,
 };
 use crate::composition::supergraph::install::InstallSupergraph;
-use crate::composition::{CompositionError, CompositionSuccess};
+use crate::composition::{
+    CompositionError, CompositionSubgraphAdded, CompositionSubgraphRemoved, CompositionSuccess,
+};
 use crate::utils::effect::exec::TokioCommand;
 use crate::utils::effect::install::InstallBinary;
 use crate::utils::effect::read_file::FsReadFile;
@@ -134,7 +136,6 @@ async fn run_lsp(client_config: StudioClientConfig, lsp_opts: LspOpts) -> RoverR
                 .await?;
 
         // Spin up Runner
-        //TODO: We need to implement some kind of From that takes us from a LazilyResolved -> FullyResolvedSubgraphs
         let mut stream = Runner::default()
             .setup_subgraph_watchers(
                 lazily_resolved_supergraph_config.subgraphs().clone(),
@@ -210,28 +211,22 @@ async fn run_lsp(client_config: StudioClientConfig, lsp_opts: LspOpts) -> RoverR
                         .publish_diagnostics(supergraph_yaml_url.clone(), vec![diagnostic])
                         .await;
                 }
+                CompositionEvent::SubgraphAdded(CompositionSubgraphAdded {
+                    name,
+                    schema_source,
+                }) => {
+                    debug!("Subgraph {} added", name);
+                    language_server.add_subgraph(name, schema_source).await;
+                }
+                CompositionEvent::SubgraphRemoved(CompositionSubgraphRemoved { name }) => {
+                    debug!("Subgraph {} removed", name);
+                    language_server.remove_subgraph(&name).await;
+                }
             }
         }
         Ok(())
     });
 
-    // Event::SubgraphAdded {
-    //     subgraph_name,
-    //     schema_source,
-    // } => {
-    //     debug!("Subgraph {} added", subgraph_name);
-    //     language_server
-    //         .add_subgraph(subgraph_name, schema_source)
-    //         .await;
-    // }
-    // Event::SubgraphRemoved { subgraph_name } => {
-    //     debug!("Subgraph {} removed", subgraph_name);
-    //     language_server.remove_subgraph(&subgraph_name).await;
-    // }
-    // Event::StartedWatchingSubgraph(watcher) => {
-    //     debug!("Started watching subgraph {watcher:?}"); // TODO: hand off between real-time and on-save
-    // }
-
     let stdin = tokio::io::stdin();
     let stdout = tokio::io::stdout();
     let server = Server::new(stdin, stdout, socket);

src/composition/events/mod.rs

@@ -1,4 +1,6 @@
-use super::{CompositionError, CompositionSuccess};
+use super::{
+    CompositionError, CompositionSubgraphAdded, CompositionSubgraphRemoved, CompositionSuccess,
+};
 
 /// Events emitted from composition
 #[derive(Debug, Eq, PartialEq)]
@@ -10,4 +12,8 @@ pub enum CompositionEvent {
     Success(CompositionSuccess),
     /// Composition errored
     Error(CompositionError),
+    /// Subgraph Added
+    SubgraphAdded(CompositionSubgraphAdded),
+    /// SubgraphRemoved
+    SubgraphRemoved(CompositionSubgraphRemoved),
 }

src/composition/mod.rs

@@ -1,5 +1,6 @@
 use std::fmt::Debug;
 
+use apollo_federation_types::config::SchemaSource;
 use apollo_federation_types::{
     config::FederationVersion,
     rover::{BuildErrors, BuildHint},
@@ -37,3 +38,14 @@ pub enum CompositionError {
     #[error("Encountered {} while trying to build a supergraph.", .source.length_string())]
     Build { source: BuildErrors },
 }
+
+#[derive(Debug, Eq, PartialEq)]
+pub struct CompositionSubgraphAdded {
+    pub(crate) name: String,
+    pub(crate) schema_source: SchemaSource,
+}
+
+#[derive(Debug, Eq, PartialEq)]
+pub struct CompositionSubgraphRemoved {
+    pub(crate) name: String,
+}

src/composition/supergraph/config/resolve/mod.rs

@@ -259,8 +259,8 @@ impl FullyResolvedSubgraphs {
     }
 
     /// Used to upsert a fully resolved subgraph into this object's definitions
-    pub fn upsert_subgraph(&mut self, name: String, schema: String) {
-        self.subgraphs.insert(name, schema);
+    pub fn upsert_subgraph(&mut self, name: String, schema: String) -> bool {
+        self.subgraphs.insert(name, schema).is_none()
     }
 
     /// Removes a subgraph from this object's definitions

src/composition/watchers/composition.rs

@@ -1,3 +1,4 @@
+use apollo_federation_types::config::SchemaSource::Sdl;
 use apollo_federation_types::config::SupergraphConfig;
 use buildstructor::Builder;
 use camino::Utf8PathBuf;
@@ -7,6 +8,7 @@ use tap::TapFallible;
 use tokio::{sync::mpsc::UnboundedSender, task::AbortHandle};
 use tokio_stream::StreamExt;
 
+use crate::composition::{CompositionSubgraphAdded, CompositionSubgraphRemoved};
 use crate::{
     composition::{
         events::CompositionEvent,
@@ -53,11 +55,25 @@ where
                         SubgraphEvent::SubgraphChanged(subgraph_schema_changed) => {
                             let name = subgraph_schema_changed.name();
                             let sdl = subgraph_schema_changed.sdl();
-                            subgraphs.upsert_subgraph(name.to_string(), sdl.to_string());
+                            if subgraphs.upsert_subgraph(name.to_string(), sdl.to_string()) {
+                                let _ = sender
+                                    .send(CompositionEvent::SubgraphAdded(
+                                        CompositionSubgraphAdded {
+                                            name: name.clone(),
+                                            schema_source: Sdl { sdl: sdl.clone() },
+                                        },
+                                    ))
+                                    .tap_err(|err| tracing::error!("{:?}", err));
+                            }
                         }
                         SubgraphEvent::SubgraphRemoved(subgraph_removed) => {
                             let name = subgraph_removed.name();
                             subgraphs.remove_subgraph(name);
+                            let _ = sender
+                                .send(CompositionEvent::SubgraphRemoved(
+                                    CompositionSubgraphRemoved { name: name.clone() },
+                                ))
+                                .tap_err(|err| tracing::error!("{:?}", err));
                         }
                     }
 

src/composition/watchers/subgraphs.rs

@@ -137,7 +137,7 @@ impl SubtaskHandleStream for SubgraphWatchers {
             // Wait for supergraph diff events received from the input stream.
             while let Some(diff) = input.next().await {
                 // If we detect additional diffs, start a new subgraph subtask.
-                // Adding the abort handle to the currentl collection of handles.
+                // Adding the abort handle to the current collection of handles.
                 for (subgraph_name, subgraph_config) in diff.added() {
                     if let Ok(subgraph_watcher) = SubgraphWatcher::from_schema_source(
                         subgraph_config.schema.clone(),