: dispatcher_(dispatcher), random_generator_(std::make_unique<Random::RandomGeneratorImpl>()),

      extension_(extension) {

  ENVOY_LOG(debug, "reverse_tunnel: creating socket manager with stats integration.");

  const std::string& dispatcher_name = dispatcher_.name();

  if (dispatcher_name != kMainThreadDispatcherName) {

    absl::WriterMutexLock lock(UpstreamSocketManager::socket_manager_lock);

    UpstreamSocketManager::socket_managers_.push_back(this);

    ENVOY_LOG(debug, "reverse_tunnel: registered socket manager for dispatcher: {}",

              dispatcher_name);

  } else {

    ENVOY_LOG(debug, "reverse_tunnel: skipping socket manager registration for main thread");

  }

}

                                                    const std::string& cluster_id) {

  absl::WriterMutexLock wlock(UpstreamSocketManager::socket_manager_lock);

  UpstreamSocketManager* target_socket_manager = this;

  const std::string source_worker = this->dispatcher_.name();

  int min_node_count = target_socket_manager->node_to_conn_count_map_[node_id];

  for (UpstreamSocketManager* socket_manager : socket_managers_) {

    int node_count = socket_manager->node_to_conn_count_map_[node_id];

    if (node_count < min_node_count) {

      target_socket_manager = socket_manager;

      min_node_count = node_count;

    }

  }

  const std::string dest_worker = target_socket_manager->dispatcher_.name();

  if (source_worker != dest_worker) {

    ENVOY_LOG(info,

              "reverse_tunnel: Rebalancing socket from worker {} to worker {} with min "

              "count {} for node {} cluster {}",

              source_worker, dest_worker, target_socket_manager->node_to_conn_count_map_[node_id],

              node_id, cluster_id);

  }

  target_socket_manager->node_to_conn_count_map_[node_id]++;

  ENVOY_LOG(debug, "reverse_tunnel: Incremented count for node {}: {}", node_id,

            target_socket_manager->node_to_conn_count_map_[node_id]);

  return *target_socket_manager;

}

                                                  const std::chrono::seconds& ping_interval) {

  dispatcher_.post(

      [this, node_id, cluster_id, ping_interval, socket = std::move(socket)]() mutable -> void {

        this->addConnectionSocket(node_id, cluster_id, std::move(socket), ping_interval,

                                  true /* rebalanced */);

      });

}

                                                bool rebalanced) {

  if (!rebalanced) {

    UpstreamSocketManager& target_manager = pickLeastLoadedSocketManager(node_id, cluster_id);

    if (&target_manager != this) {

      ENVOY_LOG(debug,

                "reverse_tunnel: Rebalancing socket to a different worker thread for node: "

                "{} cluster: {}",

                node_id, cluster_id);

      target_manager.handoffSocketToWorker(node_id, cluster_id, std::move(socket), ping_interval);

      return;

    }

  }

  ENVOY_LOG(debug, "reverse_tunnel: adding connection for node: {}, cluster: {}.", node_id,

            cluster_id);

  if (node_id.empty() || cluster_id.empty()) {

    ENVOY_LOG(error,

              "reverse_tunnel: node_id or cluster_id cannot be empty. node: '{}', cluster: '{}'.",

              node_id, cluster_id);

    return;

  }

  const int fd = socket->ioHandle().fdDoNotUse();

  const std::string& connectionKey = socket->connectionInfoProvider().localAddress()->asString();

  ENVOY_LOG(debug, "reverse_tunnel: adding socket with FD: {} for node: {}, cluster: {}.", fd,

            node_id, cluster_id);

  ENVOY_LOG(trace, "reverse_tunnel: adding mapping node {} -> cluster {}.", node_id, cluster_id);

  if (node_to_cluster_map_.find(node_id) == node_to_cluster_map_.end()) {

    node_to_cluster_map_[node_id] = cluster_id;

    cluster_to_node_info_map_[cluster_id].nodes.push_back(node_id);

  }

  fd_to_node_map_[fd] = node_id;

  fd_to_cluster_map_[fd] = cluster_id;

  node_to_active_fd_count_[node_id]++;

  fd_to_timer_map_[fd] = dispatcher_.createTimer([this, fd]() { onPingTimeout(fd); });

  accepted_reverse_connections_[node_id].push_back(std::move(socket));

  fd_to_socket_it_map_[fd] = std::prev(accepted_reverse_connections_[node_id].end());

  Network::ConnectionSocketPtr& socket_ref = accepted_reverse_connections_[node_id].back();

  if (auto extension = getUpstreamExtension()) {

    extension->updateConnectionStats(node_id, cluster_id, true /* increment */,

                                     tenant_isolation_enabled_);

    ENVOY_LOG(debug, "reverse_tunnel: updated stats registry for node '{}' cluster '{}'.", node_id,

              cluster_id);

  }

  fd_to_event_map_[fd] = dispatcher_.createFileEvent(

      fd,

      [this, &socket_ref](uint32_t events) {

        ASSERT(events == Event::FileReadyType::Read);

        onPingResponse(socket_ref->ioHandle());

        return absl::OkStatus();

      },

      Event::FileTriggerType::Edge, Event::FileReadyType::Read);

  if (ping_interval_ == std::chrono::seconds::zero()) {

    ping_interval_ = ping_interval;

  }

  fd_to_ping_send_timer_map_[fd] =

      dispatcher_.createTimer([this, fd]() { sendPingForConnection(fd); });

  fd_to_ping_send_timer_map_[fd]->enableTimer(

      std::chrono::milliseconds(pingIntervalWithJitterMs()));

  ENVOY_LOG(debug, "reverse_tunnel: added socket to maps. node: {} connection key: {} fd: {}.",

            node_id, connectionKey, fd);

}

UpstreamSocketManager::getConnectionSocket(const std::string& node_id) {

  ENVOY_LOG(debug, "reverse_tunnel: getConnectionSocket() called with node_id: {}.", node_id);

  if (node_to_cluster_map_.find(node_id) == node_to_cluster_map_.end()) {

    ENVOY_LOG(error, "reverse_tunnel: cluster to node mapping changed for node: {}.", node_id);

    return nullptr;

  }

  const std::string& cluster_id = node_to_cluster_map_[node_id];

  ENVOY_LOG(debug, "reverse_tunnel: looking for socket. node: {} cluster: {}.", node_id,

            cluster_id);

  auto node_sockets_it = accepted_reverse_connections_.find(node_id);

  if (node_sockets_it == accepted_reverse_connections_.end() || node_sockets_it->second.empty()) {

    ENVOY_LOG(debug, "reverse_tunnel: no available sockets for node: {}.", node_id);

    return nullptr;

  }

  ENVOY_LOG(trace, "reverse_tunnel: found {} sockets for node: {}.", node_sockets_it->second.size(),

            node_id);

  Network::ConnectionSocketPtr socket(std::move(node_sockets_it->second.front()));

  node_sockets_it->second.pop_front();

  const int fd = socket->ioHandle().fdDoNotUse();

  const std::string& remoteConnectionKey =

      socket->connectionInfoProvider().remoteAddress()->asString();

  ENVOY_LOG(debug,

            "reverse_tunnel: reverse connection socket found. fd: {} connection key: {} "

            "node: {} cluster: {}.",

            fd, remoteConnectionKey, node_id, cluster_id);

  fd_to_event_map_.erase(fd);

  fd_to_timer_map_.erase(fd);

  fd_to_ping_send_timer_map_.erase(fd);

  fd_to_socket_it_map_.erase(fd);

  return socket;

}

std::string UpstreamSocketManager::getNodeWithSocket(const std::string& key) {

  ENVOY_LOG(trace, "reverse_tunnel: getNodeWithSocket() called with key: {}.", key);

  auto cluster_it = cluster_to_node_info_map_.find(key);

  if (cluster_it != cluster_to_node_info_map_.end() && !cluster_it->second.nodes.empty()) {

    auto& cluster_info = cluster_it->second;

    const auto& nodes = cluster_info.nodes;

    const std::string& selected_node = nodes[cluster_info.round_robin_index % nodes.size()];

    cluster_info.round_robin_index = (cluster_info.round_robin_index + 1) % nodes.size();

    ENVOY_LOG(debug, "reverse_tunnel: key '{}' is a cluster ID; returning node {} via round-robin.",

              key, selected_node);

    return selected_node;

  }

  ENVOY_LOG(trace, "reverse_tunnel: key '{}' treated as node ID; returning as-is.", key);

  return key;

}

bool UpstreamSocketManager::hasAnySocketsForNode(const std::string& node_id) {

  auto it = node_to_active_fd_count_.find(node_id);

  return it != node_to_active_fd_count_.end() && it->second > 0;

}

void UpstreamSocketManager::markSocketDead(const int fd) {

  ENVOY_LOG(trace, "reverse_tunnel: markSocketDead called for fd {}.", fd);

  auto node_it = fd_to_node_map_.find(fd);

  if (node_it == fd_to_node_map_.end()) {

    ENVOY_LOG(warn, "reverse_tunnel: fd {} not found in fd_to_node_map_.", fd);

    return;

  }

  const std::string node_id = node_it->second;

  std::string cluster_id;

  auto cluster_it = fd_to_cluster_map_.find(fd);

  if (cluster_it == fd_to_cluster_map_.end()) {

    ENVOY_LOG(warn, "reverse_tunnel: fd {} not found in fd_to_cluster_map_.", fd);

    auto node_cluster_it = node_to_cluster_map_.find(node_id);

    if (node_cluster_it != node_to_cluster_map_.end()) {

      cluster_id = node_cluster_it->second;

    }

  } else {

    cluster_id = cluster_it->second;

  }

  ENVOY_LOG(debug, "reverse_tunnel: found node '{}' cluster '{}' for fd: {}", node_id, cluster_id,

            fd);

  fd_to_node_map_.erase(fd);

  fd_to_cluster_map_.erase(fd);

  auto count_it = node_to_active_fd_count_.find(node_id);

  if (count_it != node_to_active_fd_count_.end()) {

    ASSERT(count_it->second > 0);

    if (--count_it->second == 0) {

      node_to_active_fd_count_.erase(count_it);

    }

  }

  auto socket_it = fd_to_socket_it_map_.find(fd);

  if (socket_it != fd_to_socket_it_map_.end()) {

    ENVOY_LOG(debug, "reverse_tunnel: marking idle socket dead. node: {} cluster: {} fd: {}.",

              node_id, cluster_id, fd);

    ::shutdown(fd, SHUT_RDWR);

    accepted_reverse_connections_[node_id].erase(socket_it->second);

    fd_to_socket_it_map_.erase(socket_it);

    fd_to_event_map_.erase(fd);

    fd_to_timer_map_.erase(fd);

    fd_to_ping_send_timer_map_.erase(fd);

  } else {

    ENVOY_LOG(debug, "reverse_tunnel: marking used socket dead. node: {} cluster: {} fd: {}.",

              node_id, cluster_id, fd);

  }

  if (auto extension = getUpstreamExtension()) {

    extension->updateConnectionStats(node_id, cluster_id, false /* decrement */,

                                     tenant_isolation_enabled_);

    extension->reportDisconnection(node_id, cluster_id);

    ENVOY_LOG(trace, "reverse_tunnel: decremented stats registry for node '{}' cluster '{}'.",

              node_id, cluster_id);

  }

  if (!hasAnySocketsForNode(node_id)) {

    ENVOY_LOG(debug,

              "reverse_tunnel: node '{}' has no remaining sockets, cleaning up cluster mappings.",

              node_id);

    cleanStaleNodeEntry(node_id);

  } else {

    ENVOY_LOG(trace, "reverse_tunnel: node '{}' still has remaining sockets, keeping in maps.",

              node_id);

  }

}

void UpstreamSocketManager::cleanStaleNodeEntry(const std::string& node_id) {

  if (accepted_reverse_connections_.find(node_id) != accepted_reverse_connections_.end() &&

      accepted_reverse_connections_[node_id].size() > 0) {

    ENVOY_LOG(trace, "reverse_tunnel: found {} active sockets for node {}.",

              accepted_reverse_connections_[node_id].size(), node_id);

    return;

  }

  ENVOY_LOG(debug, "reverse_tunnel: cleaning stale node entry for node {}.", node_id);

  const auto& node_itr = node_to_cluster_map_.find(node_id);

  if (node_itr != node_to_cluster_map_.end()) {

    const auto& cluster_itr = cluster_to_node_info_map_.find(node_itr->second);

    if (cluster_itr != cluster_to_node_info_map_.end()) {

      auto& nodes = cluster_itr->second.nodes;

      const auto& node_entry_itr = find(nodes.begin(), nodes.end(), node_id);

      if (node_entry_itr != nodes.end()) {

        ENVOY_LOG(trace, "reverse_tunnel: removing stale node {} from cluster {}.", node_id,

                  cluster_itr->first);

        nodes.erase(node_entry_itr);

        if (nodes.empty()) {

          ENVOY_LOG(trace, "reverse_tunnel: removing empty cluster {}.", cluster_itr->first);

          cluster_to_node_info_map_.erase(cluster_itr);

        }

      }

    }

    node_to_cluster_map_.erase(node_itr);

  }

  accepted_reverse_connections_.erase(node_id);

}

void UpstreamSocketManager::onPingResponse(Network::IoHandle& io_handle) {

  const int fd = io_handle.fdDoNotUse();

  Buffer::OwnedImpl buffer;

  const auto ping_size =

      ::Envoy::Extensions::Bootstrap::ReverseConnection::ReverseConnectionUtility::PING_MESSAGE

          .size();

  Api::IoCallUint64Result result = io_handle.read(buffer, absl::make_optional(ping_size));

  if (!result.ok()) {

    ENVOY_LOG(debug, "reverse_tunnel: Read error on FD: {}: error - {}", fd,

              result.err_->getErrorDetails());

    markSocketDead(fd);

    return;

  }

  if (result.return_value_ == 0) {

    ENVOY_LOG(debug, "reverse_tunnel: FD: {}: reverse connection closed", fd);

    markSocketDead(fd);

    return;

  }

  if (result.return_value_ < ping_size) {

  const char* data = static_cast<const char*>(buffer.linearize(ping_size));

  absl::string_view view{data, static_cast<size_t>(ping_size)};

  if (!::Envoy::Extensions::Bootstrap::ReverseConnection::ReverseConnectionUtility::isPingMessage(

          view)) {

    ENVOY_LOG(debug, "reverse_tunnel: response is not RPING. fd: {}.", fd);

    onPingTimeout(fd);

    return;

  }

  ENVOY_LOG(trace, "reverse_tunnel: received ping response. fd: {}.", fd);

  fd_to_timer_map_[fd]->disableTimer();

  fd_to_miss_count_.erase(fd);

  rearmPingSendTimer(fd);

}

void UpstreamSocketManager::sendPingForConnection(int fd) {

  auto node_it = fd_to_node_map_.find(fd);

  if (node_it == fd_to_node_map_.end()) {

    ENVOY_LOG(debug, "reverse_tunnel: sendPingForConnection: fd {} not found in fd_to_node_map_.",

              fd);

    return;

  }

  const std::string& node_id = node_it->second;

  auto socket_it = fd_to_socket_it_map_.find(fd);

  if (socket_it == fd_to_socket_it_map_.end()) {

  Network::ConnectionSocket* socket_ptr = socket_it->second->get();

  auto buffer = ::Envoy::Extensions::Bootstrap::ReverseConnection::ReverseConnectionUtility::

      createPingResponse();

  auto ping_response_timeout = ping_interval_ / 2;

  fd_to_timer_map_[fd]->enableTimer(ping_response_timeout);

  while (buffer->length() > 0) {

    Api::IoCallUint64Result result = socket_ptr->ioHandle().write(*buffer);

    ENVOY_LOG(trace, "reverse_tunnel: node:{} FD:{}: sending ping request. return_value: {}",

              node_id, fd, result.return_value_);

    if (result.return_value_ == 0) {

      ENVOY_LOG(trace, "reverse_tunnel: node:{} FD:{}: sending ping rc {}, error - {}", node_id, fd,

                result.return_value_, result.err_->getErrorDetails());

      if (result.err_->getErrorCode() != Api::IoError::IoErrorCode::Again) {

        ENVOY_LOG(error, "reverse_tunnel: node:{} FD:{}: failed to send ping", node_id, fd);

        markSocketDead(fd);

        return;

      }

    }

  }

}

void UpstreamSocketManager::onPingTimeout(const int fd) {

  ENVOY_LOG(debug, "reverse_tunnel: ping timeout or invalid ping. fd: {}.", fd);

  const uint32_t misses = ++fd_to_miss_count_[fd];

  ENVOY_LOG(trace, "reverse_tunnel: miss count {}. fd: {}.", misses, fd);

  if (misses >= miss_threshold_) {

    ENVOY_LOG(debug, "reverse_tunnel: fd {} exceeded miss threshold {}; marking dead.", fd,

              miss_threshold_);

    fd_to_miss_count_.erase(fd);

    markSocketDead(fd);

  } else {

    rearmPingSendTimer(fd);

  }

}

uint64_t UpstreamSocketManager::pingIntervalWithJitterMs() {

  uint64_t interval_ms = static_cast<uint64_t>(ping_interval_.count()) * 1000;

  constexpr uint64_t jitter_percent = 15;

  uint64_t jitter_mod = jitter_percent * interval_ms / 100;

  if (jitter_mod > 0) {

    interval_ms += random_generator_->random() % jitter_mod;

  }

  return interval_ms;

}

void UpstreamSocketManager::rearmPingSendTimer(int fd) {

  auto send_it = fd_to_ping_send_timer_map_.find(fd);

  if (send_it != fd_to_ping_send_timer_map_.end()) {

    send_it->second->enableTimer(std::chrono::milliseconds(pingIntervalWithJitterMs()));

  }

}

UpstreamSocketManager::~UpstreamSocketManager() {

  ENVOY_LOG(debug, "reverse_tunnel: destructor called.");

  for (auto& [fd, event] : fd_to_event_map_) {

    ENVOY_LOG(trace, "reverse_tunnel: cleaning up file event. fd: {}.", fd);

    event.reset(); // This will cancel the file event.

  }

  fd_to_event_map_.clear();

  for (auto& [fd, timer] : fd_to_timer_map_) {

    ENVOY_LOG(trace, "reverse_tunnel: cleaning up timeout timer. fd: {}.", fd);

    timer.reset();

  }

  fd_to_timer_map_.clear();

  for (auto& [fd, timer] : fd_to_ping_send_timer_map_) {

    ENVOY_LOG(trace, "reverse_tunnel: cleaning up send timer. fd: {}.", fd);

    timer.reset();

  }

  fd_to_ping_send_timer_map_.clear();

  std::vector<int> fds_to_cleanup;

  for (const auto& [fd, node_id] : fd_to_node_map_) {

    fds_to_cleanup.push_back(fd);

  }

  for (int fd : fds_to_cleanup) {

    ENVOY_LOG(trace, "reverse_tunnel: marking socket dead in destructor. fd: {}.", fd);

    markSocketDead(fd);

  }

  fd_to_node_map_.clear();

  fd_to_cluster_map_.clear();

  fd_to_socket_it_map_.clear();

  absl::WriterMutexLock lock(UpstreamSocketManager::socket_manager_lock);

  auto it = std::find(socket_managers_.begin(), socket_managers_.end(), this);

  if (it != socket_managers_.end()) {

    socket_managers_.erase(it);

  }

}