#pragma once

#include <chrono>

#include <cstdint>

#include <functional>

#include <memory>

#include <string>

#include "envoy/extensions/filters/http/upstream_codec/v3/upstream_codec.pb.h"

#include "envoy/extensions/filters/http/upstream_codec/v3/upstream_codec.pb.validate.h"

#include "envoy/http/codec.h"

#include "envoy/http/codes.h"

#include "envoy/http/conn_pool.h"

#include "envoy/http/filter.h"

#include "envoy/stats/scope.h"

#include "envoy/tcp/conn_pool.h"

#include "source/common/buffer/watermark_buffer.h"

#include "source/common/common/cleanup.h"

#include "source/common/common/hash.h"

#include "source/common/common/hex.h"

#include "source/common/common/linked_object.h"

#include "source/common/common/logger.h"

#include "source/common/config/well_known_names.h"

#include "source/common/http/filter_manager.h"

#include "source/common/router/upstream_to_downstream_impl_base.h"

#include "source/common/stream_info/stream_info_impl.h"

#include "source/common/tracing/null_span_impl.h"

#include "source/extensions/filters/http/common/factory_base.h"

namespace Envoy {

namespace Router {

class GenericUpstream;

class GenericConnectionPoolCallbacks;

class RouterFilterInterface;

class UpstreamRequest;

class UpstreamRequestFilterManagerCallbacks;

class UpstreamFilterManager;

class UpstreamCodecFilter;

/* The Upstream request is the base class for forwarding HTTP upstream.

 *

 * On the new request path, payload (headers/body/metadata/data) still arrives via

 * the accept[X]fromRouter functions. Said data is immediately passed off to the

 * UpstreamFilterManager, which passes each item through the filter chain until

 * it arrives at the last filter in the chain, the UpstreamCodecFilter. If the upstream

 * stream is not established, the UpstreamCodecFilter returns StopAllIteration, and the

 * FilterManager will buffer data, using watermarks to push back to the router

 * filter if buffers become overrun. When an upstream connection is established,

 * the UpstreamCodecFilter will send data upstream.

 *

 * On the new response path, payload arrives from upstream via the UpstreamCodecFilter's

 * CodecBridge. It is passed off directly to the FilterManager, traverses the

 * filter chain, and completion is signaled via the

 * UpstreamRequestFilterManagerCallbacks's encode[X] functions. These somewhat

 * confusingly pass through the UpstreamRequest's legacy decode[X] functions

 * (required due to the UpstreamToDownstream interface, but will be renamed once

 * the classic mode is deprecated), and are finally passed to the router via the

 * RouterFilterInterface onUpstream[X] functions.

 *

 * There is some required communication between the UpstreamRequest and

 * UpstreamCodecFilter. This is accomplished via the UpstreamStreamFilterCallbacks

 * interface, with the UpstreamFilterManager acting as intermediary.

 *

 */

class UpstreamRequest : public Logger::Loggable<Logger::Id::router>,

                        public UpstreamToDownstreamImplBase,

                        public LinkedObject<UpstreamRequest>,

                        public GenericConnectionPoolCallbacks,

                        public Event::DeferredDeletable {

public:

  UpstreamRequest(RouterFilterInterface& parent, std::unique_ptr<GenericConnPool>&& conn_pool,

                  bool can_send_early_data, bool can_use_http3, bool enable_half_close);

  ~UpstreamRequest() override;

  // To be called from the destructor, or prior to deferred delete.

  void cleanUp();

  virtual void acceptHeadersFromRouter(bool end_stream);

  virtual void acceptDataFromRouter(Buffer::Instance& data, bool end_stream);

  void acceptTrailersFromRouter(Http::RequestTrailerMap& trailers);

  void acceptMetadataFromRouter(Http::MetadataMapPtr&& metadata_map_ptr);

  virtual void resetStream();

  void setupPerTryTimeout();

  void maybeEndDecode(bool end_stream);

  void onUpstreamHostSelected(Upstream::HostDescriptionConstSharedPtr host, bool pool_success);

  // Http::StreamDecoder

  void decodeData(Buffer::Instance& data, bool end_stream) override;

  void decodeMetadata(Http::MetadataMapPtr&& metadata_map) override;

  // UpstreamToDownstream (Http::ResponseDecoder)

  void decode1xxHeaders(Http::ResponseHeaderMapPtr&& headers) override;

  void decodeHeaders(Http::ResponseHeaderMapPtr&& headers, bool end_stream) override;

  void decodeTrailers(Http::ResponseTrailerMapPtr&& trailers) override;

  void dumpState(std::ostream& os, int indent_level) const override;

  // UpstreamToDownstream (Http::StreamCallbacks)

  void onResetStream(Http::StreamResetReason reason,

                     absl::string_view transport_failure_reason) override;

  // UpstreamToDownstream

  const Route& route() const override;

  OptRef<const Network::Connection> connection() const override;

  void disableDataFromDownstreamForFlowControl();

  void enableDataFromDownstreamForFlowControl();

  // GenericConnPool

  void onPoolFailure(ConnectionPool::PoolFailureReason reason,

                     absl::string_view transport_failure_reason,

                     Upstream::HostDescriptionConstSharedPtr host) override;

  void onPoolReady(std::unique_ptr<GenericUpstream>&& upstream,

                   Upstream::HostDescriptionConstSharedPtr host,

                   const Network::ConnectionInfoProvider& address_provider,

                   StreamInfo::StreamInfo& info, absl::optional<Http::Protocol> protocol) override;

  UpstreamToDownstream& upstreamToDownstream() override;

  void clearRequestEncoder();

  void onStreamMaxDurationReached();

  // Either disable upstream reading immediately or defer it and keep tracking

  // of how many read disabling has happened.

  void readDisableOrDefer(bool disable);

  // Called upon receiving the first response headers from the upstream. And

  // applies read disabling to it if there is any pending read disabling.

  void maybeHandleDeferredReadDisable();

  struct DownstreamWatermarkManager : public Http::DownstreamWatermarkCallbacks {

    // Http::DownstreamWatermarkCallbacks

    void onBelowWriteBufferLowWatermark() override;

    void onAboveWriteBufferHighWatermark() override;

    UpstreamRequest& parent_;

  };

  void readEnable();

  // Getters and setters

  // Exposes streamInfo for the upstream stream.

  // Disable per-try timeouts for websocket upgrades after successful handshake

  void disablePerTryTimeoutForWebsocketUpgrade();

private:

  friend class UpstreamFilterManager;

  friend class UpstreamCodecFilter;

  friend class UpstreamRequestFilterManagerCallbacks;

  // Records the latency from when the upstream request was first created to

  // when the pool callback fires. This latency can be useful to track excessive

  // queuing.

  void recordConnectionPoolCallbackLatency();

  void resetPerTryIdleTimer();

  void onPerTryTimeout();

  void onPerTryIdleTimeout();

  void upstreamLog(AccessLog::AccessLogType access_log_type);

  void resetUpstreamLogFlushTimer();

  RouterFilterInterface& parent_;

  std::unique_ptr<GenericConnPool> conn_pool_;

  Event::TimerPtr per_try_timeout_;

  Event::TimerPtr per_try_idle_timeout_;

  std::unique_ptr<GenericUpstream> upstream_;

  absl::optional<Http::StreamResetReason> deferred_reset_reason_;

  Upstream::HostDescriptionConstSharedPtr upstream_host_;

  DownstreamWatermarkManager downstream_watermark_manager_{*this};

  Tracing::SpanPtr span_;

  StreamInfo::StreamInfoImpl stream_info_;

  const MonotonicTime start_time_;

  // This is wrapped in an optional, since we want to avoid computing zero size headers when in

  // reality we just didn't get a response back.

  absl::optional<uint64_t> response_headers_size_;

  absl::optional<size_t> response_headers_count_;

  // Copies of upstream headers/trailers. These are only set if upstream

  // access logging is configured.

  Http::ResponseHeaderMapPtr upstream_headers_;

  Http::ResponseTrailerMapPtr upstream_trailers_;

  OptRef<UpstreamToDownstream> upstream_interface_;

  std::list<Http::UpstreamCallbacks*> upstream_callbacks_;

  Event::TimerPtr max_stream_duration_timer_;

  // Per-stream access log flush duration. This timer is enabled once when the stream is created

  // and will log to all access logs once per trigger.

  Event::TimerPtr upstream_log_flush_timer_;

  std::unique_ptr<UpstreamRequestFilterManagerCallbacks> filter_manager_callbacks_;

  std::unique_ptr<Http::FilterManager> filter_manager_;

  // The number of outstanding readDisable to be called with parameter value true.

  // When downstream send buffers get above high watermark before response headers arrive, we

  // increment this counter instead of immediately calling readDisable on upstream stream. This is

  // to avoid the upstream request from being spuriously retried or reset because of upstream

  // timeouts while upstream stream is readDisabled by downstream but the response has actually

  // arrived from upstream. See https://github.com/envoyproxy/envoy/issues/25901. During the

  // deferring period, if the downstream buffer gets below low watermark, this counter gets

  // decremented. Once the response headers arrive, call readDisable the number of times as the

  // remaining value of this counter.

  size_t deferred_read_disabling_count_{0};

  // Keep small members (bools and enums) at the end of class, to reduce alignment overhead.

  // Tracks the number of times the flow of data from downstream has been disabled.

  uint32_t downstream_data_disabled_{};

  bool upstream_canary_ : 1;

  bool router_sent_end_stream_ : 1;

  bool encode_trailers_ : 1;

  bool retried_ : 1;

  bool awaiting_headers_ : 1;

  bool outlier_detection_timeout_recorded_ : 1;

  // Tracks whether we deferred a per try timeout because the downstream request

  // had not been completed yet.

  bool create_per_try_timeout_on_request_complete_ : 1;

  // True if the CONNECT headers have been sent but proxying payload is paused

  // waiting for response headers.

  bool paused_for_connect_ : 1;

  bool paused_for_websocket_ : 1;

  bool reset_stream_ : 1;

  // Sentinel to indicate if timeout budget tracking is configured for the cluster,

  // and if so, if the per-try histogram should record a value.

  bool record_timeout_budget_ : 1;

  // Track if one time clean up has been performed.

  bool cleaned_up_ : 1;

  bool had_upstream_ : 1;

  Http::ConnectionPool::Instance::StreamOptions stream_options_;

  bool grpc_rq_success_deferred_ : 1;

  bool enable_half_close_ : 1;

};

class UpstreamRequestFilterManagerCallbacks : public Http::FilterManagerCallbacks,

                                              public Event::DeferredDeletable,

                                              public Http::UpstreamStreamFilterCallbacks {

public:

  UpstreamRequestFilterManagerCallbacks(UpstreamRequest& upstream_request)

  Http::RequestHeaderMapOptRef requestHeaders() override;

  Http::RequestTrailerMapOptRef requestTrailers() override;

  // If the filter manager determines a decoder filter has available, tell

  // the router to resume the flow of data from downstream.

  // If the filter manager determines a decoder filter has too much data, tell

  // the router to stop the flow of data from downstream.

  // These functions are delegated to the downstream HCM/FM

  OptRef<const Tracing::Config> tracingConfig() const override;

  const ScopeTrackedObject& scope() override;

  Tracing::Span& activeSpan() override;

  void resetStream(Http::StreamResetReason reset_reason,

                   absl::string_view transport_failure_reason) override;

  Upstream::ClusterInfoConstSharedPtr clusterInfo() override;

  Http::Http1StreamEncoderOptionsOptRef http1StreamEncoderOptions() override;

  // Intentional no-op functions.

  // Upgrade filter chains not supported.

  // Unsupported functions.

  // Http::UpstreamStreamFilterCallbacks

  void disableRouteTimeoutForWebsocketUpgrade() override;

  void disablePerTryTimeoutForWebsocketUpgrade() override;

  Http::RequestTrailerMapPtr trailers_;

  Http::ResponseHeaderMapPtr informational_headers_;

  Http::ResponseHeaderMapPtr response_headers_;

  Http::ResponseTrailerMapPtr response_trailers_;

  UpstreamRequest& upstream_request_;

};

} // namespace Router

} // namespace Envoy