feat: rover dev listens for .env file changes and hot reloads router (#2984)

<!--
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.

-->
Fix for: #2750

- Adds a file watcher for `.env`
- Rereads the config file and hot reloads the router on `.env` file
changes

### Testing
**Setup**:
`router.yaml`
```
supergraph:
  listen: 0.0.0.0:${env.ROUTER_PORT}
```

`.env`
```
ROUTER_PORT=4000
```

**Run `rover dev`**
```
/rover dev --graph-ref my-test@current --router-config router.yaml
retrieving subgraphs remotely from my-test@current
merging supergraph schema files
starting a session with the 'starwars' subgraph
composing supergraph with Federation 2.12.2
==> Watching router.yaml for changes
==> Attempting to start router at http://localhost:4000.
==> Health check exposed at http://127.0.0.1:8088/health
WARN: Connector debugging is enabled, this may expose sensitive information.
==> Your supergraph is running! head to http://localhost:4000 to query your supergraph
```

**Change the `.env` file by updating the router port to 4002**
```
DotEnvWatcher: .env file changed
==> Router config updated.
supergraph:
  listen: 0.0.0.0:4002

==> Health check exposed at http://127.0.0.1:8088/health
==> Your supergraph is running! head to http://localhost:4002 to query your supergraph
```

Author: DMallare

src/command/dev/router/run.rs

@@ -20,7 +20,10 @@ use super::{
     config::{ReadRouterConfigError, RouterAddress, RunRouterConfig, remote::RemoteRouterConfig},
     hot_reload::{HotReloadEvent, HotReloadWatcher, RouterUpdateEvent},
     install::{InstallRouter, InstallRouterError},
-    watchers::router_config::RouterConfigWatcher,
+    watchers::{
+        dot_env::{DotEnvEvent, DotEnvReload, DotEnvWatcher},
+        router_config::RouterConfigWatcher,
+    },
 };
 use crate::{
     RoverError,
@@ -380,7 +383,7 @@ impl RunRouter<state::Watch> {
     {
         tracing::info!("Watching for subgraph changes");
         let (router_config_updates, config_watcher_subtask) =
-            if let Some(config_path) = self.state.config_path {
+            if let Some(config_path) = self.state.config_path.clone() {
                 let config_watcher = RouterConfigWatcher::new(FileWatcher::new(config_path));
                 let (events, subtask): (UnboundedReceiverStream<RouterUpdateEvent>, _) =
                     Subtask::new(config_watcher);
@@ -389,6 +392,16 @@ impl RunRouter<state::Watch> {
                 (None, None)
             };
 
+        let (dot_env_updates, dot_env_watcher_subtask) = if dotenvy::dotenv().is_ok() {
+            tracing::info!("Watching for .env changes");
+            let dot_env_watcher = DotEnvWatcher::new();
+            let (events, subtask): (UnboundedReceiverStream<DotEnvEvent>, _) =
+                Subtask::new(dot_env_watcher);
+            (Some(events), Some(subtask))
+        } else {
+            (None, None)
+        };
+
         let composition_messages =
             tokio_stream::StreamExt::filter_map(composition_messages, |event| match event {
                 CompositionEvent::Error(CompositionError::Build { source, .. }) => {
@@ -421,11 +434,33 @@ impl RunRouter<state::Watch> {
 
         let (hot_reload_events, hot_reload_subtask): (UnboundedReceiverStream<HotReloadEvent>, _) =
             Subtask::new(hot_reload_watcher);
-        let router_config_updates = router_config_updates
-            .map(move |stream| stream.boxed())
-            .unwrap_or_else(|| stream::empty().boxed());
-        let router_updates =
-            tokio_stream::StreamExt::merge(router_config_updates, composition_messages);
+
+        let mut streams: Vec<_> = vec![];
+        streams.push(composition_messages.boxed());
+
+        if let Some(stream) = router_config_updates {
+            streams.push(stream.boxed());
+        }
+
+        if let (Some(dot_env_event_stream), Some(config_path)) =
+            (dot_env_updates, self.state.config_path.clone())
+        {
+            // Emit RouterUpdateEvent::ConfigChanged for each .env change
+            let dot_env_reload = DotEnvReload {
+                router_config_path: config_path,
+            };
+            let (router_config_updates, dot_env_reload_subtask): (
+                UnboundedReceiverStream<RouterUpdateEvent>,
+                _,
+            ) = Subtask::new(dot_env_reload);
+            dot_env_reload_subtask.run(
+                dot_env_event_stream.boxed(),
+                Some(self.state.cancellation_token.clone()),
+            );
+            streams.push(router_config_updates.boxed());
+        }
+
+        let router_updates = stream::select_all(streams).boxed();
 
         SubtaskRunStream::run(
             hot_reload_subtask,
@@ -436,6 +471,9 @@ impl RunRouter<state::Watch> {
         if let Some(subtask) = config_watcher_subtask {
             subtask.run(Some(self.state.cancellation_token.clone()))
         }
+        if let Some(subtask) = dot_env_watcher_subtask {
+            subtask.run(Some(self.state.cancellation_token.clone()))
+        }
 
         RunRouter {
             state: state::Abort {

src/command/dev/router/watchers/dot_env.rs

@@ -0,0 +1,102 @@
+use camino::Utf8PathBuf;
+use dotenvy::dotenv_override;
+use futures::{StreamExt, stream::BoxStream};
+use rover_std::Fs;
+use tap::TapFallible;
+use tokio::sync::mpsc::UnboundedSender;
+use tokio_util::sync::CancellationToken;
+
+use crate::{
+    command::dev::router::{
+        config::RouterConfig, hot_reload::RouterUpdateEvent, watchers::file::FileWatcher,
+    },
+    subtask::{SubtaskHandleStream, SubtaskHandleUnit},
+};
+
+pub struct DotEnvWatcher {
+    file_watcher: FileWatcher,
+}
+
+#[derive(Debug)]
+pub enum DotEnvEvent {
+    DotEnvChanged,
+}
+
+impl DotEnvWatcher {
+    pub fn new() -> Self {
+        let dot_env_path = dotenvy::dotenv()
+            .ok()
+            .and_then(|p| Utf8PathBuf::from_path_buf(p).ok())
+            .unwrap_or_else(|| Utf8PathBuf::from(".env"));
+
+        Self {
+            file_watcher: FileWatcher::new(dot_env_path),
+        }
+    }
+}
+
+impl SubtaskHandleUnit for DotEnvWatcher {
+    type Output = DotEnvEvent;
+    fn handle(
+        self,
+        sender: tokio::sync::mpsc::UnboundedSender<Self::Output>,
+        cancellation_token: Option<CancellationToken>,
+    ) {
+        let cancellation_token = cancellation_token.unwrap_or_default();
+        tokio::spawn(async move {
+            cancellation_token
+                .run_until_cancelled(async move {
+                    // Emit a DotEnvEvent::Changed event which will trigger a router config reload
+                    while let Some(_env_contents) = self.file_watcher.clone().watch().next().await {
+                        let _ = sender
+                            .send(DotEnvEvent::DotEnvChanged)
+                            .tap_err(|err| tracing::error!("{:?}", err));
+                    }
+                })
+                .await;
+        });
+    }
+}
+
+pub struct DotEnvReload {
+    pub router_config_path: Utf8PathBuf,
+}
+
+impl SubtaskHandleStream for DotEnvReload {
+    type Input = DotEnvEvent;
+
+    type Output = RouterUpdateEvent;
+
+    fn handle(
+        self,
+        sender: UnboundedSender<Self::Output>,
+        mut input: BoxStream<'static, Self::Input>,
+        cancellation_token: Option<CancellationToken>,
+    ) {
+        let cancellation_token = cancellation_token.unwrap_or_default();
+        tokio::spawn(async move {
+            cancellation_token
+                .run_until_cancelled(async move {
+                    while let Some(_evt) = input.next().await {
+                        dotenv_override().ok();
+                        match Fs::read_file(self.router_config_path.clone()) {
+                            Ok(config_contents) => {
+                                let _ = sender
+                                    .send(RouterUpdateEvent::ConfigChanged {
+                                        config: RouterConfig::new(config_contents),
+                                    })
+                                    .tap_err(|err| tracing::error!("{:?}", err));
+                            }
+                            Err(err) => {
+                                tracing::error!(
+                                    "Could not read router config after .env change: {:?}",
+                                    err
+                                );
+                            }
+                        }
+                    }
+                })
+                .await;
+        });
+    }
+}

src/command/dev/router/watchers/mod.rs

@@ -1,2 +1,3 @@
+pub mod dot_env;
 pub mod file;
 pub mod router_config;