#include "common/router/scoped_rds.h"

#include <memory>

#include "envoy/admin/v3/config_dump.pb.h"
#include "envoy/api/v2/scoped_route.pb.h"
#include "envoy/config/core/v3/config_source.pb.h"
#include "envoy/config/route/v3/scoped_route.pb.h"
#include "envoy/extensions/filters/network/http_connection_manager/v3/http_connection_manager.pb.h"
#include "envoy/service/discovery/v3/discovery.pb.h"

#include "common/common/assert.h"
#include "common/common/cleanup.h"
#include "common/common/logger.h"
#include "common/common/utility.h"
#include "common/config/api_version.h"
#include "common/config/resource_name.h"
#include "common/config/version_converter.h"
#include "common/init/manager_impl.h"
#include "common/init/watcher_impl.h"
#include "common/router/rds_impl.h"
#include "common/router/scoped_config_impl.h"

#include "absl/strings/str_join.h"

// Types are deeply nested under Envoy::Config::ConfigProvider; use 'using-directives' across all
// ConfigProvider related types for consistency.
using Envoy::Config::ConfigProvider;
using Envoy::Config::ConfigProviderInstanceType;
using Envoy::Config::ConfigProviderManager;
using Envoy::Config::ConfigProviderPtr;
using Envoy::Config::ScopedResume;

namespace Envoy {
namespace Router {
namespace ScopedRoutesConfigProviderUtil {
ConfigProviderPtr create(
    const envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager&
        config,
    Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager,
    const std::string& stat_prefix, ConfigProviderManager& scoped_routes_config_provider_manager) {
  ASSERT(config.route_specifier_case() ==
         envoy::extensions::filters::network::http_connection_manager::v3::HttpConnectionManager::
             RouteSpecifierCase::kScopedRoutes);

  switch (config.scoped_routes().config_specifier_case()) {
  case envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
      ConfigSpecifierCase::kScopedRouteConfigurationsList: {
    const envoy::extensions::filters::network::http_connection_manager::v3::
        ScopedRouteConfigurationsList& scoped_route_list =
            config.scoped_routes().scoped_route_configurations_list();
    return scoped_routes_config_provider_manager.createStaticConfigProvider(
        RepeatedPtrUtil::convertToConstMessagePtrContainer<
            envoy::config::route::v3::ScopedRouteConfiguration,
            ProtobufTypes::ConstMessagePtrVector>(scoped_route_list.scoped_route_configurations()),
        factory_context,
        ScopedRoutesConfigProviderManagerOptArg(config.scoped_routes().name(),
                                                config.scoped_routes().rds_config_source(),
                                                config.scoped_routes().scope_key_builder()));
  }
  case envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
      ConfigSpecifierCase::kScopedRds:
    return scoped_routes_config_provider_manager.createXdsConfigProvider(
        config.scoped_routes().scoped_rds(), factory_context, init_manager, stat_prefix,
        ScopedRoutesConfigProviderManagerOptArg(config.scoped_routes().name(),
                                                config.scoped_routes().rds_config_source(),
                                                config.scoped_routes().scope_key_builder()));
  default:
    // Proto validation enforces that is not reached.
    NOT_REACHED_GCOVR_EXCL_LINE;
  }
}

} // namespace ScopedRoutesConfigProviderUtil

InlineScopedRoutesConfigProvider::InlineScopedRoutesConfigProvider(
    ProtobufTypes::ConstMessagePtrVector&& config_protos, std::string name,
    Server::Configuration::ServerFactoryContext& factory_context,
    ScopedRoutesConfigProviderManager& config_provider_manager,
    envoy::config::core::v3::ConfigSource rds_config_source,
    envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::ScopeKeyBuilder
        scope_key_builder)
    : Envoy::Config::ImmutableConfigProviderBase(factory_context, config_provider_manager,
                                                 ConfigProviderInstanceType::Inline,
                                                 ConfigProvider::ApiType::Delta),
      name_(std::move(name)),
      config_(std::make_shared<ScopedConfigImpl>(std::move(scope_key_builder))),
      config_protos_(std::make_move_iterator(config_protos.begin()),
                     std::make_move_iterator(config_protos.end())),
      rds_config_source_(std::move(rds_config_source)) {}

ScopedRdsConfigSubscription::ScopedRdsConfigSubscription(
    const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRds& scoped_rds,
    const uint64_t manager_identifier, const std::string& name,
    const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
        ScopeKeyBuilder& scope_key_builder,
    Server::Configuration::ServerFactoryContext& factory_context, const std::string& stat_prefix,
    envoy::config::core::v3::ConfigSource rds_config_source,
    RouteConfigProviderManager& route_config_provider_manager,
    ScopedRoutesConfigProviderManager& config_provider_manager)
    : DeltaConfigSubscriptionInstance("SRDS", manager_identifier, config_provider_manager,
                                      factory_context),
      Envoy::Config::SubscriptionBase<envoy::config::route::v3::ScopedRouteConfiguration>(
          scoped_rds.scoped_rds_config_source().resource_api_version(),
          factory_context.messageValidationContext().dynamicValidationVisitor(), "name"),
      factory_context_(factory_context), name_(name), scope_key_builder_(scope_key_builder),
      scope_(factory_context.scope().createScope(stat_prefix + "scoped_rds." + name + ".")),
      stats_({ALL_SCOPED_RDS_STATS(POOL_COUNTER(*scope_), POOL_GAUGE(*scope_))}),
      rds_config_source_(std::move(rds_config_source)), stat_prefix_(stat_prefix),
      route_config_provider_manager_(route_config_provider_manager) {
  const auto resource_name = getResourceName();
  subscription_ =
      factory_context.clusterManager().subscriptionFactory().subscriptionFromConfigSource(
          scoped_rds.scoped_rds_config_source(), Grpc::Common::typeUrl(resource_name), *scope_,
          *this, resource_decoder_);

  initialize([scope_key_builder]() -> Envoy::Config::ConfigProvider::ConfigConstSharedPtr {
    return std::make_shared<ScopedConfigImpl>(
        envoy::extensions::filters::network::http_connection_manager::v3::ScopedRoutes::
            ScopeKeyBuilder(scope_key_builder));
  });
}

// Constructor for RdsRouteConfigProviderHelper when scope is eager loading.
// Initialize RdsRouteConfigProvider by default.
ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::RdsRouteConfigProviderHelper(
    ScopedRdsConfigSubscription& parent, std::string scope_name,
    envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
    Init::Manager& init_manager)
    : parent_(parent), scope_name_(scope_name), on_demand_(false) {
  initRdsConfigProvider(rds, init_manager);
}

// Constructor for RdsRouteConfigProviderHelper when scope is on demand.
// Leave the RdsRouteConfigProvider uninitialized.
ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::RdsRouteConfigProviderHelper(
    ScopedRdsConfigSubscription& parent, std::string scope_name)
    : parent_(parent), scope_name_(scope_name), on_demand_(true) {
  parent_.stats_.on_demand_scopes_.inc();
}

// When on demand callback is received from main thread, there are 4 cases.
// 1. Scope is not found, post a scope not found callback back to worker thread.
// 2. Scope is found but route provider has not been initialized, create route provider.
// 3. After route provider has been initialized, if RouteConfiguration has been fetched,
// post scope found callback to worker thread.
// 4. After route provider has been initialized, if RouteConfiguration is null,
// cache the callback and wait for RouteConfiguration to come.
void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::addOnDemandUpdateCallback(
    std::function<void()> callback) {
  // If RouteConfiguration has been initialized, run the callback to continue in filter chain,
  // otherwise cache it and wait for the route table to be initialized. If RouteConfiguration hasn't
  // been initialized, routeConfig() return a shared_ptr to NullConfigImpl. The name of
  // NullConfigImpl is an empty string.
  if (route_provider_ != nullptr && !routeConfig()->name().empty()) {
    callback();
    return;
  }
  on_demand_update_callbacks_.push_back(callback);
  // Initialize the rds provider if it has not been initialized. There is potential race here
  // because other worker threads may also post callback to on demand update the RouteConfiguration
  // associated with this scope. If rds provider has been initialized, just wait for
  // RouteConfiguration to be updated.
  maybeInitRdsConfigProvider();
}

void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::runOnDemandUpdateCallback() {
  for (auto& callback : on_demand_update_callbacks_) {
    callback();
  }
  on_demand_update_callbacks_.clear();
}

void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::initRdsConfigProvider(
    envoy::extensions::filters::network::http_connection_manager::v3::Rds& rds,
    Init::Manager& init_manager) {
  route_provider_ = std::dynamic_pointer_cast<RdsRouteConfigProviderImpl>(
      parent_.route_config_provider_manager_.createRdsRouteConfigProvider(
          rds, parent_.factory_context_, parent_.stat_prefix_, init_manager));

  rds_update_callback_handle_ = route_provider_->subscription().addUpdateCallback([this]() {
    // Subscribe to RDS update.
    parent_.onRdsConfigUpdate(scope_name_, route_provider_->subscription());
  });
  parent_.stats_.active_scopes_.inc();
}

void ScopedRdsConfigSubscription::RdsRouteConfigProviderHelper::maybeInitRdsConfigProvider() {
  // If the route provider have been initialized, return and wait for rds config update.
  if (route_provider_ != nullptr) {
    return;
  }

  // Create a init_manager to create a rds provider.
  // No transitive warming dependency here because only on demand update reach this point.
  std::unique_ptr<Init::ManagerImpl> srds_init_mgr =
      std::make_unique<Init::ManagerImpl>(fmt::format("SRDS on demand init manager."));
  std::unique_ptr<Cleanup> srds_initialization_continuation =
      std::make_unique<Cleanup>([this, &srds_init_mgr] {
        Init::WatcherImpl noop_watcher(
            fmt::format("SRDS on demand ConfigUpdate watcher: {}", scope_name_),
            []() { /*Do nothing.*/ });
        srds_init_mgr->initialize(noop_watcher);
      });
  // Create route provider.
  envoy::extensions::filters::network::http_connection_manager::v3::Rds rds;
  rds.mutable_config_source()->MergeFrom(parent_.rds_config_source_);
  rds.set_route_config_name(
      parent_.scoped_route_map_[scope_name_]->configProto().route_configuration_name());
  initRdsConfigProvider(rds, *srds_init_mgr);
  ENVOY_LOG(debug, fmt::format("Scope on demand update: {}", scope_name_));
  // If RouteConfiguration hasn't been initialized, routeConfig() return a shared_ptr to
  // NullConfigImpl. The name of NullConfigImpl is an empty string.
  if (routeConfig()->name().empty()) {
    return;
  }
  // If RouteConfiguration has been initialized, apply update to all the threads.
  parent_.onRdsConfigUpdate(scope_name_, route_provider_->subscription());
}

bool ScopedRdsConfigSubscription::addOrUpdateScopes(
    const std::vector<Envoy::Config::DecodedResourceRef>& resources, Init::Manager& init_manager,
    const std::string& version_info) {
  bool any_applied = false;
  envoy::extensions::filters::network::http_connection_manager::v3::Rds rds;
  rds.mutable_config_source()->MergeFrom(rds_config_source_);
  std::vector<ScopedRouteInfoConstSharedPtr> updated_scopes;
  for (const auto& resource : resources) {
    // Explicit copy so that we can std::move later.
    envoy::config::route::v3::ScopedRouteConfiguration scoped_route_config =
        dynamic_cast<const envoy::config::route::v3::ScopedRouteConfiguration&>(
            resource.get().resource());
    const std::string scope_name = scoped_route_config.name();
    rds.set_route_config_name(scoped_route_config.route_configuration_name());
    std::unique_ptr<RdsRouteConfigProviderHelper> rds_config_provider_helper;
    std::shared_ptr<ScopedRouteInfo> scoped_route_info = nullptr;
    if (scoped_route_config.on_demand() == false) {
      // For default scopes, create a rds helper with rds provider initialized.
      rds_config_provider_helper =
          std::make_unique<RdsRouteConfigProviderHelper>(*this, scope_name, rds, init_manager);
      scoped_route_info = std::make_shared<ScopedRouteInfo>(
          std::move(scoped_route_config), rds_config_provider_helper->routeConfig());
    } else {
      // For on demand scopes, create a rds helper with rds provider uninitialized.
      rds_config_provider_helper =
          std::make_unique<RdsRouteConfigProviderHelper>(*this, scope_name);
      // scope_route_info->routeConfig() will be nullptr, because RouteConfiguration is not loaded.
      scoped_route_info =
          std::make_shared<ScopedRouteInfo>(std::move(scoped_route_config), nullptr);
    }
    route_provider_by_scope_[scope_name] = std::move(rds_config_provider_helper);
    scope_name_by_hash_[scoped_route_info->scopeKey().hash()] = scoped_route_info->scopeName();
    scoped_route_map_[scoped_route_info->scopeName()] = scoped_route_info;
    updated_scopes.push_back(scoped_route_info);
    any_applied = true;
    ENVOY_LOG(debug, "srds: queueing add/update of scoped_route '{}', version: {}",
              scoped_route_info->scopeName(), version_info);
  }

  // scoped_route_info of both eager loading and on demand scopes will be propagated to work
  // threads. Upon a scoped RouteConfiguration miss, if the scope exists, an on demand update
  // callback will be posted to main thread.
  if (!updated_scopes.empty()) {
    applyConfigUpdate([updated_scopes](ConfigProvider::ConfigConstSharedPtr config)
                          -> ConfigProvider::ConfigConstSharedPtr {
      auto* thread_local_scoped_config =
          const_cast<ScopedConfigImpl*>(static_cast<const ScopedConfigImpl*>(config.get()));
      thread_local_scoped_config->addOrUpdateRoutingScopes(updated_scopes);
      return config;
    });
  }
  return any_applied;
}

std::list<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelperPtr>
ScopedRdsConfigSubscription::removeScopes(
    const Protobuf::RepeatedPtrField<std::string>& scope_names, const std::string& version_info) {
  std::list<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelperPtr>
      to_be_removed_rds_providers;
  std::vector<std::string> removed_scope_names;
  for (const auto& scope_name : scope_names) {
    auto iter = scoped_route_map_.find(scope_name);
    if (iter != scoped_route_map_.end()) {
      auto rds_config_provider_helper_iter = route_provider_by_scope_.find(scope_name);
      if (rds_config_provider_helper_iter != route_provider_by_scope_.end()) {
        to_be_removed_rds_providers.emplace_back(
            std::move(rds_config_provider_helper_iter->second));
        route_provider_by_scope_.erase(rds_config_provider_helper_iter);
      }
      scope_name_by_hash_.erase(iter->second->scopeKey().hash());
      scoped_route_map_.erase(iter);
      removed_scope_names.push_back(scope_name);
      ENVOY_LOG(debug, "srds: queueing removal of scoped route '{}', version: {}", scope_name,
                version_info);
    }
  }
  if (!removed_scope_names.empty()) {
    applyConfigUpdate([removed_scope_names](ConfigProvider::ConfigConstSharedPtr config)
                          -> ConfigProvider::ConfigConstSharedPtr {
      auto* thread_local_scoped_config =
          const_cast<ScopedConfigImpl*>(static_cast<const ScopedConfigImpl*>(config.get()));
      thread_local_scoped_config->removeRoutingScopes(removed_scope_names);
      return config;
    });
  }
  return to_be_removed_rds_providers;
}

void ScopedRdsConfigSubscription::onConfigUpdate(
    const std::vector<Envoy::Config::DecodedResourceRef>& added_resources,
    const Protobuf::RepeatedPtrField<std::string>& removed_resources,
    const std::string& version_info) {
  // NOTE: deletes are done before adds/updates.
  absl::flat_hash_map<std::string, ScopedRouteInfoConstSharedPtr> to_be_removed_scopes;
  // Destruction of resume_rds will lift the floodgate for new RDS subscriptions.
  // Note in the case of partial acceptance, accepted RDS subscriptions should be started
  // despite of any error.
  ScopedResume resume_rds;
  // If new route config sources come after the local init manager's initialize() been
  // called, the init manager can't accept new targets. Instead we use a local override which will
  // start new subscriptions but not wait on them to be ready.
  std::unique_ptr<Init::ManagerImpl> srds_init_mgr;
  // NOTE: This should be defined after srds_init_mgr and resume_rds, as it depends on the
  // srds_init_mgr, and we want a single RDS discovery request to be sent to management
  // server.
  std::unique_ptr<Cleanup> srds_initialization_continuation;
  ASSERT(localInitManager().state() > Init::Manager::State::Uninitialized);
  const auto type_urls =
      Envoy::Config::getAllVersionTypeUrls<envoy::config::route::v3::RouteConfiguration>();
  // Pause RDS to not send a burst of RDS requests until we start all the new subscriptions.
  // In the case that localInitManager is uninitialized, RDS is already paused
  // either by Server init or LDS init.
  if (factory_context_.clusterManager().adsMux()) {
    resume_rds = factory_context_.clusterManager().adsMux()->pause(type_urls);
  }
  // if local init manager is initialized, the parent init manager may have gone away.
  if (localInitManager().state() == Init::Manager::State::Initialized) {
    srds_init_mgr =
        std::make_unique<Init::ManagerImpl>(fmt::format("SRDS {}:{}", name_, version_info));
    srds_initialization_continuation =
        std::make_unique<Cleanup>([this, &srds_init_mgr, version_info] {
          // For new RDS subscriptions created after listener warming up, we don't wait for them to
          // warm up.
          Init::WatcherImpl noop_watcher(
              // Note: we just throw it away.
              fmt::format("SRDS ConfigUpdate watcher {}:{}", name_, version_info),
              []() { /*Do nothing.*/ });
          srds_init_mgr->initialize(noop_watcher);
        });
  }

  std::string exception_msg;
  Protobuf::RepeatedPtrField<std::string> clean_removed_resources =
      detectUpdateConflictAndCleanupRemoved(added_resources, removed_resources, exception_msg);
  if (!exception_msg.empty()) {
    throw EnvoyException(fmt::format("Error adding/updating scoped route(s): {}", exception_msg));
  }

  // Do not delete RDS config providers just yet, in case the to be deleted RDS subscriptions could
  // be reused by some to be added scopes.
  std::list<ScopedRdsConfigSubscription::RdsRouteConfigProviderHelperPtr>
      to_be_removed_rds_providers = removeScopes(clean_removed_resources, version_info);

  bool any_applied =
      addOrUpdateScopes(added_resources,
                        (srds_init_mgr == nullptr ? localInitManager() : *srds_init_mgr),
                        version_info) ||
      !to_be_removed_rds_providers.empty();
  ConfigSubscriptionCommonBase::onConfigUpdate();
  if (any_applied) {
    setLastConfigInfo(absl::optional<LastConfigInfo>({absl::nullopt, version_info}));
  }
  stats_.all_scopes_.set(scoped_route_map_.size());
  stats_.config_reload_.inc();
}

void ScopedRdsConfigSubscription::onRdsConfigUpdate(const std::string& scope_name,
                                                    RdsRouteConfigSubscription& rds_subscription) {
  auto iter = scoped_route_map_.find(scope_name);
  ASSERT(iter != scoped_route_map_.end(),
         fmt::format("trying to update route config for non-existing scope {}", scope_name));
  auto new_scoped_route_info = std::make_shared<ScopedRouteInfo>(
      envoy::config::route::v3::ScopedRouteConfiguration(iter->second->configProto()),
      std::make_shared<ConfigImpl>(
          rds_subscription.routeConfigUpdate()->routeConfiguration(), factory_context_,
          factory_context_.messageValidationContext().dynamicValidationVisitor(), false));
  applyConfigUpdate([new_scoped_route_info](ConfigProvider::ConfigConstSharedPtr config)
                        -> ConfigProvider::ConfigConstSharedPtr {
    auto* thread_local_scoped_config =
        const_cast<ScopedConfigImpl*>(static_cast<const ScopedConfigImpl*>(config.get()));
    thread_local_scoped_config->addOrUpdateRoutingScopes({new_scoped_route_info});
    return config;
  });
  // The data plane may wait for the route configuration to come back.
  route_provider_by_scope_[scope_name]->runOnDemandUpdateCallback();
}

// TODO(stevenzzzz): see issue #7508, consider generalizing this function as it overlaps with
// CdsApiImpl::onConfigUpdate.
void ScopedRdsConfigSubscription::onConfigUpdate(
    const std::vector<Envoy::Config::DecodedResourceRef>& resources,
    const std::string& version_info) {
  Protobuf::RepeatedPtrField<std::string> to_remove_repeated;
  for (const auto& scoped_route : scoped_route_map_) {
    *to_remove_repeated.Add() = scoped_route.first;
  }
  onConfigUpdate(resources, to_remove_repeated, version_info);
}

Protobuf::RepeatedPtrField<std::string>
ScopedRdsConfigSubscription::detectUpdateConflictAndCleanupRemoved(
    const std::vector<Envoy::Config::DecodedResourceRef>& resources,
    const Protobuf::RepeatedPtrField<std::string>& removed_resources, std::string& exception_msg) {
  Protobuf::RepeatedPtrField<std::string> clean_removed_resources;
  // All the scope names to be removed or updated.
  absl::flat_hash_set<std::string> updated_or_removed_scopes;
  for (const std::string& removed_resource : removed_resources) {
    updated_or_removed_scopes.insert(removed_resource);
  }
  for (const auto& resource : resources) {
    const auto& scoped_route =
        dynamic_cast<const envoy::config::route::v3::ScopedRouteConfiguration&>(
            resource.get().resource());
    updated_or_removed_scopes.insert(scoped_route.name());
  }

  absl::flat_hash_map<uint64_t, std::string> scope_name_by_hash = scope_name_by_hash_;
  absl::erase_if(scope_name_by_hash, [&updated_or_removed_scopes](const auto& key_name) {
    auto const& [key, name] = key_name;
    UNREFERENCED_PARAMETER(key);
    return updated_or_removed_scopes.contains(name);
  });
  absl::flat_hash_map<std::string, envoy::config::route::v3::ScopedRouteConfiguration>
      scoped_routes;
  for (const auto& resource : resources) {
    // Throws (thus rejects all) on any error.
    const auto& scoped_route =
        dynamic_cast<const envoy::config::route::v3::ScopedRouteConfiguration&>(
            resource.get().resource());
    const std::string& scope_name = scoped_route.name();
    auto scope_config_inserted = scoped_routes.try_emplace(scope_name, std::move(scoped_route));
    if (!scope_config_inserted.second) {
      exception_msg = fmt::format("duplicate scoped route configuration '{}' found", scope_name);
      return clean_removed_resources;
    }
    const envoy::config::route::v3::ScopedRouteConfiguration& scoped_route_config =
        scope_config_inserted.first->second;
    const uint64_t key_fingerprint = MessageUtil::hash(scoped_route_config.key());
    if (!scope_name_by_hash.try_emplace(key_fingerprint, scope_name).second) {
      exception_msg =
          fmt::format("scope key conflict found, first scope is '{}', second scope is '{}'",
                      scope_name_by_hash[key_fingerprint], scope_name);
      return clean_removed_resources;
    }
  }

  // only remove resources that is not going to be updated.
  for (const std::string& removed_resource : removed_resources) {
    if (!scoped_routes.contains(removed_resource)) {
      *clean_removed_resources.Add() = removed_resource;
    }
  }
  return clean_removed_resources;
}

void ScopedRdsConfigSubscription::onDemandRdsUpdate(
    std::shared_ptr<Router::ScopeKey> scope_key, Event::Dispatcher& thread_local_dispatcher,
    Http::RouteConfigUpdatedCallback&& route_config_updated_cb,
    std::weak_ptr<Envoy::Config::ConfigSubscriptionCommonBase> weak_subscription) {
  factory_context_.dispatcher().post([this, &thread_local_dispatcher, scope_key,
                                      route_config_updated_cb, weak_subscription]() {
    // If the subscription has been destroyed, return immediately.
    if (!weak_subscription.lock()) {
      thread_local_dispatcher.post([route_config_updated_cb] { route_config_updated_cb(false); });
      return;
    }

    auto iter = scope_name_by_hash_.find(scope_key->hash());
    // Return to filter chain if we can't find the scope.
    // The scope may have been destroyed when callback reach the main thread.
    if (iter == scope_name_by_hash_.end()) {
      thread_local_dispatcher.post([route_config_updated_cb] { route_config_updated_cb(false); });
      return;
    }
    // Wrap the thread local dispatcher inside the callback.
    std::function<void()> thread_local_updated_callback = [route_config_updated_cb,
                                                           &thread_local_dispatcher]() {
      thread_local_dispatcher.post([route_config_updated_cb] { route_config_updated_cb(true); });
    };
    std::string scope_name = iter->second;
    // On demand initialization inside main thread.
    route_provider_by_scope_[scope_name]->addOnDemandUpdateCallback(thread_local_updated_callback);
  });
}

ScopedRdsConfigProvider::ScopedRdsConfigProvider(
    ScopedRdsConfigSubscriptionSharedPtr&& subscription)
    : MutableConfigProviderCommonBase(std::move(subscription), ConfigProvider::ApiType::Delta) {}

ProtobufTypes::MessagePtr ScopedRoutesConfigProviderManager::dumpConfigs() const {
  auto config_dump = std::make_unique<envoy::admin::v3::ScopedRoutesConfigDump>();
  for (const auto& element : configSubscriptions()) {
    auto subscription = element.second.lock();
    ASSERT(subscription);

    if (subscription->configInfo()) {
      auto* dynamic_config = config_dump->mutable_dynamic_scoped_route_configs()->Add();
      dynamic_config->set_version_info(subscription->configInfo().value().last_config_version_);
      const ScopedRdsConfigSubscription* typed_subscription =
          static_cast<ScopedRdsConfigSubscription*>(subscription.get());
      dynamic_config->set_name(typed_subscription->name());
      const ScopedRouteMap& scoped_route_map = typed_subscription->scopedRouteMap();
      for (const auto& it : scoped_route_map) {
        dynamic_config->mutable_scoped_route_configs()->Add()->PackFrom(
            API_RECOVER_ORIGINAL(it.second->configProto()));
      }
      TimestampUtil::systemClockToTimestamp(subscription->lastUpdated(),
                                            *dynamic_config->mutable_last_updated());
    }
  }

  for (const auto& provider : immutableConfigProviders(ConfigProviderInstanceType::Inline)) {
    const auto protos_info =
        provider->configProtoInfoVector<envoy::config::route::v3::ScopedRouteConfiguration>();
    ASSERT(protos_info != absl::nullopt);
    auto* inline_config = config_dump->mutable_inline_scoped_route_configs()->Add();
    inline_config->set_name(static_cast<InlineScopedRoutesConfigProvider*>(provider)->name());
    for (const auto& config_proto : protos_info.value().config_protos_) {
      inline_config->mutable_scoped_route_configs()->Add()->PackFrom(
          API_RECOVER_ORIGINAL(*config_proto));
    }
    TimestampUtil::systemClockToTimestamp(provider->lastUpdated(),
                                          *inline_config->mutable_last_updated());
  }

  return config_dump;
}

ConfigProviderPtr ScopedRoutesConfigProviderManager::createXdsConfigProvider(
    const Protobuf::Message& config_source_proto,
    Server::Configuration::ServerFactoryContext& factory_context, Init::Manager& init_manager,
    const std::string& stat_prefix, const ConfigProviderManager::OptionalArg& optarg) {
  const auto& typed_optarg = static_cast<const ScopedRoutesConfigProviderManagerOptArg&>(optarg);
  ScopedRdsConfigSubscriptionSharedPtr subscription =
      ConfigProviderManagerImplBase::getSubscription<ScopedRdsConfigSubscription>(
          config_source_proto, init_manager,
          [&config_source_proto, &factory_context, &stat_prefix,
           &typed_optarg](const uint64_t manager_identifier,
                          ConfigProviderManagerImplBase& config_provider_manager)
              -> Envoy::Config::ConfigSubscriptionCommonBaseSharedPtr {
            const auto& scoped_rds_config_source = dynamic_cast<
                const envoy::extensions::filters::network::http_connection_manager::v3::ScopedRds&>(
                config_source_proto);
            return std::make_shared<ScopedRdsConfigSubscription>(
                scoped_rds_config_source, manager_identifier, typed_optarg.scoped_routes_name_,
                typed_optarg.scope_key_builder_, factory_context, stat_prefix,
                typed_optarg.rds_config_source_,
                static_cast<ScopedRoutesConfigProviderManager&>(config_provider_manager)
                    .routeConfigProviderPanager(),
                static_cast<ScopedRoutesConfigProviderManager&>(config_provider_manager));
          });

  return std::make_unique<ScopedRdsConfigProvider>(std::move(subscription));
}

ConfigProviderPtr ScopedRoutesConfigProviderManager::createStaticConfigProvider(
    ProtobufTypes::ConstMessagePtrVector&& config_protos,
    Server::Configuration::ServerFactoryContext& factory_context,
    const ConfigProviderManager::OptionalArg& optarg) {
  const auto& typed_optarg = static_cast<const ScopedRoutesConfigProviderManagerOptArg&>(optarg);
  return std::make_unique<InlineScopedRoutesConfigProvider>(
      std::move(config_protos), typed_optarg.scoped_routes_name_, factory_context, *this,
      typed_optarg.rds_config_source_, typed_optarg.scope_key_builder_);
}

} // namespace Router
} // namespace Envoy