repo_description: Build, bundle & ship your Rust WASM application to the web.
repo_topics: assets build-tool bundler compiler html rust sass scss wasm wasm-bindgen
code: use crate::{
      build::{BuildResult, BuildSystem},
      config::{
          rt::{GlobMatcher, RtcWatch},
          types::WsProtocol,
      },
      ws,
  };
  use anyhow::{Context, Result};
  use futures_util::stream::StreamExt;
  use notify::{
      EventKind, PollWatcher, RecommendedWatcher, RecursiveMode, Watcher,
      event::{MetadataKind, ModifyKind},
  };
  use notify_debouncer_full::{
      DebounceEventResult, DebouncedEvent, Debouncer, FileIdMap, new_debouncer_opt,
  };
  use std::path::Path;
  use std::{fmt::Write, path::PathBuf, sync::Arc, time::Duration};
  use tokio::{
      sync::{Mutex, broadcast, mpsc, watch},
      time::Instant,
  };
  use tokio_stream::wrappers::BroadcastStream;
  
  pub enum FsDebouncer {
      Default(Debouncer<RecommendedWatcher, FileIdMap>),
      Polling(Debouncer<PollWatcher, FileIdMap>),
  }
  
  impl FsDebouncer {
      pub fn watch(
          &mut self,
          path: impl AsRef<Path>,
          recursive_mode: RecursiveMode,
      ) -> notify::Result<()> {
          match self {
              Self::Default(deb) => deb.watch(path, recursive_mode),
              Self::Polling(deb) => deb.watch(path, recursive_mode),
          }
      }
  }
  
  /// Blacklisted path segments which are ignored by the watcher by default.
  const BLACKLIST: [&str; 2] = [".git", ".DS_Store"];
  /// The duration of time to debounce FS events.
  const DEBOUNCE_DURATION: Duration = Duration::from_millis(25);
  /// The duration of time during which watcher events will be ignored following a build.
  ///
  /// There are various OS syscalls which can trigger FS changes, even though semantically
  /// no changes were made. A notorious example which has plagued the trunk
  /// watcher implementation is `std::fs::copy`, which will trigger watcher
  /// changes indicating that file contents have been modified.
  ///
  /// Given the difficult nature of this issue, we opt for using a cooldown period. Any
  /// changes events processed within the cooldown period following a build
  /// will be ignored.
  const WATCHER_COOLDOWN: Duration = Duration::from_secs(1);
  
  /// A watch system wrapping a build system and a watcher.
  pub struct WatchSystem {
      /// The build system.
      build: Arc<Mutex<BuildSystem>>,
      /// The current vector of paths to be ignored.
      ignored_paths: GlobMatcher,
      /// A channel of FS watch events.
      watch_rx: mpsc::Receiver<DebouncedEvent>,
      /// A channel of new paths to ignore from the build system.
      ignore_rx: mpsc::Receiver<Vec<PathBuf>>,
      /// A sender to notify the end of a build.
      build_tx: mpsc::Sender<BuildResult>,
      /// A channel to receive the end of a build.
      build_rx: mpsc::Receiver<BuildResult>,
      /// The watch system used for watching the filesystem.
      _debouncer: FsDebouncer,
      /// The application shutdown channel.
      shutdown: BroadcastStream<()>,
      /// Channel to communicate with the client socket
      ws_state: Option<watch::Sender<ws::State>>,
      /// Timestamp the last build was started.
      last_build_started: Instant,
      /// An instant used to track the last build time, used to implement the watcher cooldown
      /// to avoid infinite build loops.
      ///
      /// Ok, so why is this needed? As it turns out, `std::fs::copy` will trigger
      /// `EventKind::Modify(ModifyKind::Data(_))` FS events on the file which is being copied. A
      /// build cooldown period ensures that no FS events are processed until at least a duration
      /// of `WATCHER_COOLDOWN` has elapsed since the last build.
      last_build_finished: Instant,
      /// The timestamp of the last accepted change event.
      last_change: Instant,
      /// The cooldown for the watcher. [`None`] disables the cooldown.
      watcher_cooldown: Option<Duration>,
      /// Clear the screen before each run
      clear_screen: bool,
      /// Don't send build errors to the frontend.
      no_error_reporting: bool,
  }
  
  impl WatchSystem {
      /// Create a new instance.
      pub async fn new(
          cfg: Arc<RtcWatch>,
          shutdown: broadcast::Sender<()>,
          ws_state: Option<watch::Sender<ws::State>>,
          ws_protocol: Option<WsProtocol>,
      ) -> Result<Self> {
          // Create a channel for being able to listen for new paths to ignore while running.
          let (watch_tx, watch_rx) = mpsc::channel(1);
          let (ignore_tx, ignore_rx) = mpsc::channel(1);
          let (build_tx, build_rx) = mpsc::channel(1);
  
          // Build the watcher.
          let _debouncer = build_watcher(watch_tx, cfg.paths.clone(), cfg.poll)?;
  
          // Cooldown
          let watcher_cooldown = cfg.enable_cooldown.then_some(WATCHER_COOLDOWN);
          tracing::debug!(
              "Build cooldown: {:?}",
              watcher_cooldown.map(humantime::Duration::from)
          );
  
          // Build dependencies.
          let build = Arc::new(Mutex::new(
              BuildSystem::new(cfg.build.clone(), Some(ignore_tx), ws_protocol).await?,
          ));
          Ok(Self {
              build,
              ignored_paths: cfg.ignored_paths.clone(),
              watch_rx,
              ignore_rx,
              build_rx,
              build_tx,
              _debouncer,
              shutdown: BroadcastStream::new(shutdown.subscribe()),
              ws_state,
              last_build_started: Instant::now(),
              last_build_finished: Instant::now(),
              last_change: Instant::now(),
              watcher_cooldown,
              clear_screen: cfg.clear_screen,
              no_error_reporting: cfg.no_error_reporting,
          })
      }
  
      /// Run a build.
      #[tracing::instrument(level = "trace", skip(self))]
      pub async fn build(&mut self) -> Result<()> {
          self.build.lock().await.build().await
      }
  
      /// Run the watch system, responding to events and triggering builds.
      #[tracing::instrument(level = "trace", skip(self))]
      pub async fn run(mut self) {
          loop {
              tokio::select! {
                  Some(ign) = self.ignore_rx.recv() => ign.into_iter().for_each(|ign| self.update_ignore_list(ign)),
                  Some(ev) = self.watch_rx.recv() => self.handle_watch_event(ev).await,
                  Some(build) = self.build_rx.recv() => self.build_complete(build).await,
                  _ = self.shutdown.next() => break, // Any event, even a drop, will trigger shutdown.
              }
          }
  
          tracing::debug!("watcher system has shut down");
      }
  
      #[tracing::instrument(level = "trace", skip(self))]
      async fn build_complete(&mut self, build_result: Result<(), anyhow::Error>) {
          tracing::debug!("Build reported completion");
  
          // record last finish timestamp
          self.last_build_finished = Instant::now();
  
          if let Some(tx) = &mut self.ws_state {
              match build_result {
                  Ok(()) => {
                      let _ = tx.send_replace(ws::State::Ok);
                  }
                  Err(err) => {
                      if !self.no_error_reporting {
                          let _ = tx.send_replace(ws::State::Failed {
                              reason: build_error_reason(err),
                          });
                      }
                  }
              }
          }
  
          // check we need another build
          self.check_spawn_build().await;
      }
  
      /// check if a build is active
      fn is_build_active(&self) -> bool {
          self.last_build_started > self.last_build_finished
      }
  
      /// Spawn a new build
      async fn spawn_build(&mut self) {
          self.last_build_started = Instant::now();
  
          let build = self.build.clone();
          let build_tx = self.build_tx.clone();
  
          tokio::spawn(async move {
              // run the build
              let result = build.lock().await.build().await;
              // report the result
              build_tx.send(result).await
          });
      }
  
      async fn check_spawn_build(&mut self) {
          if self.last_change <= self.last_build_started {
              tracing::trace!("No changes since the last build was started");
              return;
          }
  
          tracing::debug!("Changes since the last build was started, checking cooldown");
  
          if let Some(cooldown) = self.watcher_cooldown {
              let time_since_last_build = self
                  .last_change
                  .saturating_duration_since(self.last_build_finished);
              if time_since_last_build < cooldown {
                  tracing::debug!(
                      "Cooldown is still active: {} remaining",
                      humantime::Duration::from(cooldown - time_since_last_build)
                  );
                  return;
              }
          }
  
          if self.clear_screen {
              // This first message will not be seen if the clear screen worked.
              tracing::trace!("Clear screen is enabled, clearing the screen");
              let term = console::Term::stdout();
              if let Err(err) = term.clear_screen() {
                  tracing::error!("Unable to clear the screen due to error: #{err}");
              } else {
                  tracing::trace!("Clear screen is enabled, cleared the screen");
              }
          }
          self.spawn_build().await;
      }
  
      #[tracing::instrument(level = "trace", skip(self, event))]
      async fn handle_watch_event(&mut self, event: DebouncedEvent) {
          tracing::trace!(
              "change detected in {:?} of type {:?}",
              event.paths,
              event.kind
          );
  
          if !self.is_event_relevant(&event).await {
              tracing::trace!("Event not relevant, skipping");
              return;
          }
  
          // record time of the last accepted change
          self.last_change = Instant::now();
  
          if self.is_build_active() {
              tracing::debug!("Build is active, postponing start");
              return;
          }
  
          // Else, time to trigger a build.
          self.check_spawn_build().await;
      }
  
      async fn is_event_relevant(&self, event: &DebouncedEvent) -> bool {
          // Check each path in the event for a match.
          match event.event.kind {
              EventKind::Modify(
                  ModifyKind::Name(_)
                  | ModifyKind::Data(_)
                  | ModifyKind::Metadata(MetadataKind::WriteTime)
                  | ModifyKind::Any,
              )
              | EventKind::Create(_)
              | EventKind::Remove(_) => (),
              _ => return false,
          };
  
          for ev_path in &event.paths {
              let ev_path = match tokio::fs::canonicalize(&ev_path).await {
                  Ok(ev_path) => ev_path,
                  // Ignore errors here, as this would only take place for a resource which has
                  // been removed, which will happen for each of our dist/.stage entries.
                  Err(_) => continue,
              };
  
              // Check ignored paths.
              if self.ignored_paths.is_match(&ev_path) {
                  continue;
              }
  
              // Check blacklisted paths.
              if ev_path
                  .components()
                  .filter_map(|segment| segment.as_os_str().to_str())
                  .any(|segment| BLACKLIST.contains(&segment))
              {
                  continue; // Don't emit a notification as path is on the blacklist.
              }
  
              // If all of the above checks have passed, then we need to trigger a build.
              tracing::debug!("accepted change in {:?} of type {:?}", ev_path, event.kind);
              // But we can return early, as we don't need to check the remaining changes
              return true;
          }
  
          false
      }
  
      fn update_ignore_list(&mut self, arg_path: PathBuf) {
          let Some(path) = arg_path.to_str() else {
              tracing::warn!("could not convert {arg_path:?} to str");
              return;
          };
          let path = match globset::Glob::new(path) {
              Ok(path) => path,
              Err(err) => {
                  tracing::warn!("invalid glob: {err:?}");
                  return;
              }
          };
  
          // SAFETY: All previous patterns are valid, and so is
          // the new one being added, so this should never panic.
          #[allow(clippy::expect_used)]
          self.ignored_paths
              .add(path)
              .expect("all patterns to be valid");
      }
  }
  
  fn new_debouncer<T: Watcher>(
      watch_tx: mpsc::Sender<DebouncedEvent>,
      config: Option<notify::Config>,
  ) -> Result<Debouncer<T, FileIdMap>> {
      new_debouncer_opt::<_, T, FileIdMap>(
          DEBOUNCE_DURATION,
          None,
          move |result: DebounceEventResult| match result {
              Ok(events) => events.into_iter().for_each(|event| {
                  let _ = watch_tx.blocking_send(event);
              }),
              Err(errors) => errors
                  .into_iter()
                  .for_each(|err| tracing::warn!(error=?err, "error from filesystem watcher")),
          },
          FileIdMap::new(),
          config.unwrap_or_default(),
      )
      .context("failed to build file system watcher")
  }
  
  /// Build a FS watcher, when the watcher is dropped, it will stop watching for events.
  fn build_watcher(
      watch_tx: mpsc::Sender<DebouncedEvent>,
      paths: Vec<PathBuf>,
      poll: Option<Duration>,
  ) -> Result<FsDebouncer> {
      // Build the filesystem watcher & debouncer.
  
      if let Some(duration) = poll {
          tracing::info!(
              "Running in polling mode: {}",
              humantime::Duration::from(duration)
          );
      }
  
      let mut debouncer = match poll {
          None => FsDebouncer::Default(new_debouncer::<RecommendedWatcher>(watch_tx, None)?),
          Some(duration) => FsDebouncer::Polling(new_debouncer::<PollWatcher>(
              watch_tx,
              Some(notify::Config::default().with_poll_interval(duration)),
          )?),
      };
  
      // Create a recursive watcher on each of the given paths.
      // NOTE WELL: it is expected that all given paths are canonical. The Trunk config
      // system currently ensures that this is true for all data coming from the
      // RtcBuild/RtcWatch/RtcServe/&c runtime config objects.
      for path in paths {
          debouncer
              .watch(&path, RecursiveMode::Recursive)
              .context(format!("failed to watch {path:?} for file system changes",))?;
      }
  
      Ok(debouncer)
  }
  
  fn build_error_reason(error: anyhow::Error) -> String {
      let mut result = error.to_string();
      result.push_str("\n\n");
  
      let mut i = 0usize;
      let mut next = error.source();
      while let Some(current) = next {
          if i == 0 {
              #[allow(clippy::unwrap_used)]
              writeln!(&mut result, "Caused by:").unwrap();
          }
          #[allow(clippy::unwrap_used)]
          writeln!(&mut result, "\t{i}: {current}").unwrap();
  
          i += 1;
          next = current.source();
      }
  
      result
  }
  