text: /*
   * Copyright (c) Meta Platforms, Inc. and affiliates.
   *
   * This source code is licensed under the MIT license found in the
   * LICENSE file in the root directory of this source tree.
   */
  
  #include <folly/Synchronized.h>
  #include <condition_variable>
  #include <mutex>
  #include "watchman/InMemoryView.h"
  #include "watchman/root/Root.h"
  #include "watchman/watcher/WatcherRegistry.h"
  #include "watchman/watcher/fsevents.h"
  #include "watchman/watcher/kqueue.h"
  #include "watchman/watchman_cmd.h"
  #include "watchman/watchman_file.h"
  
  #if HAVE_FSEVENTS && defined(HAVE_KQUEUE)
  namespace watchman {
  
  class PendingEventsCond {
   public:
    /**
     * Notify that some events are pending.
     *
     * Return true if this thread should stop, false otherwise.
     */
    bool notifyOneOrStop() {
      auto lock = stop_.lock();
      if (lock->shouldStop) {
        return true;
      }
      lock->hasPending = true;
      cond_.notify_one();
      return false;
    }
  
    /**
     * Whether this thread should stop.
     */
    bool shouldStop() {
      return stop_.lock()->shouldStop;
    }
  
    /**
     * Wait for a change from a nested watcher. Return true if some events are
     * pending.
     */
    bool waitAndClear(int timeoutms) {
      auto lock = stop_.lock();
      cond_.wait_until(
          lock.as_lock(),
          std::chrono::steady_clock::now() + std::chrono::milliseconds(timeoutms),
          [&] { return lock->hasPending || lock->shouldStop; });
      return std::exchange(lock->hasPending, false);
    }
  
    /**
     * Notify all the waiting threads to stop.
     */
    void stopAll() {
      auto lock = stop_.lock();
      lock->shouldStop = true;
      cond_.notify_all();
    }
  
   private:
    struct Inner {
      bool shouldStop = false;
      bool hasPending = false;
    };
  
    folly::Synchronized<Inner, std::mutex> stop_;
    std::condition_variable cond_;
  };
  
  /**
   * Watcher that uses both kqueue and fsevents to watch a hierarchy.
   *
   * The kqueue watches are used on the root directory and all the files at the
   * root, while the fsevents one is used on the subdirectories.
   */
  class KQueueAndFSEventsWatcher : public Watcher {
   public:
    explicit KQueueAndFSEventsWatcher(
        const w_string& root_path,
        const Configuration& config);
  
    bool start(const std::shared_ptr<Root>& root) override;
  
    folly::SemiFuture<folly::Unit> flushPendingEvents() override;
  
    std::unique_ptr<DirHandle> startWatchDir(
        const std::shared_ptr<Root>& root,
        struct watchman_dir* dir,
        const char* path) override;
  
    bool startWatchFile(struct watchman_file* file) override;
  
    Watcher::ConsumeNotifyRet consumeNotify(
        const std::shared_ptr<Root>& root,
        PendingChanges& coll) override;
  
    bool waitNotify(int timeoutms) override;
    void stopThreads() override;
  
    /**
     * Force a recrawl to be injected in the stream. Used in the
     * 'debug-kqueue-and-fsevents-recrawl' command.
     */
    void injectRecrawl(w_string path);
  
   private:
    folly::Synchronized<
        std::unordered_map<w_string, std::shared_ptr<FSEventsWatcher>>>
        fseventWatchers_;
    std::shared_ptr<KQueueWatcher> kqueueWatcher_;
  
    std::shared_ptr<PendingEventsCond> pendingCondition_;
  
    folly::Synchronized<std::optional<w_string>> injectedRecrawl_;
  };
  
  KQueueAndFSEventsWatcher::KQueueAndFSEventsWatcher(
      const w_string& root_path,
      const Configuration& config)
      : Watcher("kqueue+fsevents", WATCHER_HAS_SPLIT_WATCH),
        kqueueWatcher_(std::make_shared<KQueueWatcher>(root_path, config, false)),
        pendingCondition_(std::make_shared<PendingEventsCond>()) {}
  
  namespace {
  bool startThread(
      const std::shared_ptr<Root>& root,
      const std::shared_ptr<Watcher>& watcher,
      const std::shared_ptr<PendingEventsCond>& cond) {
    std::weak_ptr<Watcher> weakWatcher(watcher);
    std::thread thr([weakWatcher, root, cond]() {
      while (true) {
        auto watcher = weakWatcher.lock();
        if (!watcher) {
          break;
        }
        if (watcher->waitNotify(86400)) {
          if (cond->notifyOneOrStop()) {
            return;
          }
        } else if (cond->shouldStop()) {
          return;
        }
      }
    });
    thr.detach();
    return true;
  }
  } // namespace
  
  bool KQueueAndFSEventsWatcher::start(const std::shared_ptr<Root>& root) {
    root->cookies.addCookieDir(root->root_path);
    return startThread(root, kqueueWatcher_, pendingCondition_);
  }
  
  folly::SemiFuture<folly::Unit> KQueueAndFSEventsWatcher::flushPendingEvents() {
    // Flush the kqueue watcher outside of the lock, because it may need to
    // change the set of watchers.
    auto kqueueFlush = kqueueWatcher_->flushPendingEvents();
    // But we know KQueueWatcher doesn't implement flushPendingEvents, so to
    // avoid having to chain the futures here, just assert.
    w_check(
        !kqueueFlush.valid(),
        "This code needs to be updated to handle KQueueWatcher implementing flushPendingEvents");
  
    auto fseventsWatchers = *fseventWatchers_.rlock();
  
    std::vector<folly::SemiFuture<folly::Unit>> futures;
    futures.reserve(fseventsWatchers.size());
    for (auto& [name, watcher] : fseventsWatchers) {
      auto future = watcher->flushPendingEvents();
      if (future.valid()) {
        futures.push_back(std::move(future));
      }
    }
    return folly::collect(futures).unit();
  }
  
  std::unique_ptr<DirHandle> KQueueAndFSEventsWatcher::startWatchDir(
      const std::shared_ptr<Root>& root,
      struct watchman_dir* dir,
      const char* path) {
    if (!dir->parent) {
      logf(DBG, "Watching root directory with kqueue\n");
      // This is the root, let's watch it with kqueue.
      kqueueWatcher_->startWatchDir(root, dir, path);
    } else if (dir->parent->getFullPath() == root->root_path) {
      auto fullPath = dir->getFullPath();
      auto wlock = fseventWatchers_.wlock();
      if (wlock->find(fullPath) == wlock->end()) {
        logf(
            DBG,
            "Creating a new FSEventsWatcher for top-level directory {}\n",
            dir->name);
        root->cookies.addCookieDir(fullPath);
        auto [it, _] = wlock->emplace(
            fullPath,
            std::make_shared<FSEventsWatcher>(
                false, root->config, std::optional(fullPath)));
        const auto& watcher = it->second;
        if (!watcher->start(root)) {
          throw std::runtime_error("couldn't start fsEvent");
        }
        if (!startThread(root, watcher, pendingCondition_)) {
          throw std::runtime_error("couldn't start fsEvent");
        }
      }
    }
  
    return openDir(path);
  }
  
  bool KQueueAndFSEventsWatcher::startWatchFile(struct watchman_file* file) {
    if (file->parent->parent == nullptr) {
      // File at the root, watch it with kqueue.
      return kqueueWatcher_->startWatchFile(file);
    }
  
    // FSEvent by default watches all the files recursively, we don't need to do
    // anything.
    return true;
  }
  
  Watcher::ConsumeNotifyRet KQueueAndFSEventsWatcher::consumeNotify(
      const std::shared_ptr<Root>& root,
      PendingChanges& coll) {
    {
      auto guard = injectedRecrawl_.wlock();
      if (guard->has_value()) {
        const auto& injectedDir = guard->value();
  
        auto now = std::chrono::system_clock::now();
        coll.add(
            injectedDir,
            now,
            W_PENDING_VIA_NOTIFY | W_PENDING_RECURSIVE | W_PENDING_IS_DESYNCED);
  
        guard->reset();
      }
    }
  
    {
      auto fseventWatches = fseventWatchers_.wlock();
      for (auto& [watchpath, fsevent] : *fseventWatches) {
        auto [cancelSelf] = fsevent->consumeNotify(root, coll);
        if (cancelSelf) {
          fsevent->stopThreads();
          root->cookies.removeCookieDir(watchpath);
          fseventWatches->erase(watchpath);
          continue;
        }
      }
    }
  
    return kqueueWatcher_->consumeNotify(root, coll);
  }
  
  bool KQueueAndFSEventsWatcher::waitNotify(int timeoutms) {
    return pendingCondition_->waitAndClear(timeoutms);
  }
  
  void KQueueAndFSEventsWatcher::stopThreads() {
    pendingCondition_->stopAll();
    {
      auto fseventWatches = fseventWatchers_.rlock();
      for (auto& [_, fsevent] : *fseventWatches) {
        fsevent->stopThreads();
      }
    }
    kqueueWatcher_->stopThreads();
  }
  
  void KQueueAndFSEventsWatcher::injectRecrawl(w_string path) {
    *injectedRecrawl_.wlock() = path;
    pendingCondition_->notifyOneOrStop();
  }
  
  namespace {
  std::shared_ptr<InMemoryView> makeKQueueAndFSEventsWatcher(
      const w_string& root_path,
      const w_string& /*fstype*/,
      const Configuration& config) {
    if (config.getBool("prefer_split_fsevents_watcher", false)) {
      return std::make_shared<InMemoryView>(
          realFileSystem,
          root_path,
          config,
          std::make_shared<KQueueAndFSEventsWatcher>(root_path, config));
    } else {
      throw std::runtime_error(
          "Not using the kqueue+fsevents watcher as the \"prefer_split_fsevents_watcher\" config isn't set");
    }
  }
  } // namespace
  
  static WatcherRegistry reg("kqueue+fsevents", makeKQueueAndFSEventsWatcher, 5);
  
  namespace {
  
  std::shared_ptr<KQueueAndFSEventsWatcher> watcherFromRoot(
      const std::shared_ptr<Root>& root) {
    auto view = std::dynamic_pointer_cast<watchman::InMemoryView>(root->view());
    if (!view) {
      return nullptr;
    }
  
    return std::dynamic_pointer_cast<KQueueAndFSEventsWatcher>(
        view->getWatcher());
  }
  
  static void cmd_debug_kqueue_and_fsevents_recrawl(
      struct watchman_client* client,
      const json_ref& args) {
    /* resolve the root */
    if (json_array_size(args) != 3) {
      send_error_response(
          client,
          "wrong number of arguments for 'debug-kqueue-and-fsevents-recrawl'");
      return;
    }
  
    auto root = resolveRoot(client, args);
  
    auto watcher = watcherFromRoot(root);
    if (!watcher) {
      send_error_response(
          client, "root is not using the kqueue+fsevents watcher");
      return;
    }
  
    /* Get the path that the recrawl should be triggered on */
    const auto& json_path = args.at(2);
    auto path = json_string_value(json_path);
    if (!path) {
      send_error_response(
          client,
          "invalid value for argument 2, expected a string naming the path to trigger a recrawl on");
    }
  
    watcher->injectRecrawl(path);
  
    auto resp = make_response();
    send_and_dispose_response(client, std::move(resp));
  }
  
  } // namespace
  
  W_CMD_REG(
      "debug-kqueue-and-fsevents-recrawl",
      cmd_debug_kqueue_and_fsevents_recrawl,
      CMD_DAEMON,
      w_cmd_realpath_root)
  
  } // namespace watchman
  
  #endif
  