defmodule Finch do
  @external_resource "README.md"
  @moduledoc "README.md"
             |> File.read!()
             |> String.split("<!-- MDOC !-->")
             |> Enum.fetch!(1)

  alias Finch.{Pool, Request, Response}
  require Finch.Pool.Manager

  use Supervisor

  @default_pool_size 50
  @default_pool_count 1

  @default_connect_timeout 5_000

  @pool_config_schema [
    protocols: [
      type: {:list, {:in, [:http1, :http2]}},
      doc: """
      The type of connections to support.

      If using `:http1` only, an HTTP1 pool without multiplexing is used. \
      If using `:http2` only, an HTTP2 pool with multiplexing is used. \
      If both are listed, then both HTTP1/HTTP2 connections are \
      supported (via ALPN), but there is no multiplexing.
      """,
      default: [:http1]
    ],
    count: [
      type: :pos_integer,
      doc: """
      How many **shards** to start for this pool key.

      - **HTTP/1**: Each shard is a `NimblePool`. HTTP/1 shards are able to re-use connections in
      the same shard and establish new ones only when necessary. A higher `:count` under moderate
      traffic scatters work so idle connections stay **per shard**, which **reduces HTTP/1
      connection reuse**. Prefer the **lowest `:count`** that still meets latency and throughput;
      raise **`count`** when you see checkout queue timeouts or heavy load on one shard. Use
      `t:pool_metrics/0`, `get_pool_status/2`, and Finch telemetry to inspect connections per shard.

      - **HTTP/2**: Each shard is a single connection process, able to multiplex requests. Shards
      register under the same registry key, so increasing `:count` spreads concurrent load across
      more processes and can relieve pressure when a **single** pool process (message handling,
      socket operations) becomes the bottleneck. Prefer the **lowest `:count`** unless one shard is
      the limit; raise **`count`** when telemetry or `get_pool_status/2` shows a shard consistently
      hot (e.g. high `in_flight_requests`).

      When `:count` > 1, **`:pool_strategy`** selects the shard per request—see [Multiple shards
      (`count` > 1) and `:pool_strategy`](#start_link/1-multiple-shards-count-1-and-pool_strategy).
      """,
      default: @default_pool_count
    ],
    size: [
      type: :pos_integer,
      doc: """
      It is the maximum number of HTTP/1 connections per pool **shard**. Connections are opened
      **lazily** up to this cap; the value is an upper bound, not a reservation. When every
      connection in that shard is busy, further requests wait in the checkout queue until one is
      returned or `:pool_timeout` (see `request/3`) is exceeded.

      This applies only to HTTP/1 pools. For HTTP/2, this setting is ignored. A single connection
      multiplexes streams per pool process; use `:count` for more HTTP/2 connections in parallel.

      Combined with `:count`, the upper bound on concurrent HTTP/1 connections to one origin
      is roughly **`count * size`**. Actual open connections may be lower.
      """,
      default: @default_pool_size
    ],
    conn_opts: [
      type: :keyword_list,
      doc: """
      These options are passed to `Mint.HTTP.connect/4` whenever a new connection is established. \
      `:mode` is not configurable as Finch must control this setting. Typically these options are \
      used to configure proxying, https settings, or connect timeouts.
      """,
      default: []
    ],
    pool_max_idle_time: [
      type: :timeout,
      doc: """
      The maximum number of milliseconds that a pool can be idle before being terminated, used only by HTTP1 pools. \
      This options is forwarded to NimblePool and it starts and idle verification cycle that may impact \
      performance if misused. For instance setting a very low timeout may lead to pool restarts. \
      For more information see NimblePool's `handle_ping/2` documentation.
      """,
      default: :infinity
    ],
    conn_max_idle_time: [
      type: :timeout,
      doc: """
      The maximum number of milliseconds an HTTP1 connection is allowed to be idle \
      before being closed during a checkout attempt.
      """,
      default: :infinity
    ],
    start_pool_metrics?: [
      type: :boolean,
      doc: "When true, pool metrics will be collected and available through `get_pool_status/2`",
      default: false
    ],
    http2: [
      type: :keyword_list,
      doc: "HTTP/2-specific options. Only relevant when `protocols` includes `:http2`.",
      default: [
        wait_for_server_settings?: false,
        ping_interval: :infinity,
        max_connection_age: :infinity,
        max_connection_age_jitter: 0
      ],
      keys: [
        wait_for_server_settings?: [
          type: :boolean,
          doc: """
          When true, the pool does not send any request until the server's SETTINGS frame \
          has been received and applied. If a request arrives before that, it fails with \
          `Finch.Error` reason `:connection_not_ready` (callers should retry). When false, \
          behaviour is unchanged and requests may be sent before SETTINGS.
          """,
          default: false
        ],
        ping_interval: [
          type: :timeout,
          default: :infinity,
          doc: """
          Interval in milliseconds between HTTP/2 PING frames sent to keep the connection \
          alive. The timer resets on any connection activity, so PINGs are only sent after \
          the connection has been idle for this duration. When set to `:infinity` (default), \
          no PINGs are sent.
          """
        ],
        max_connection_age: [
          type: :timeout,
          doc: """
          Maximum lifetime in milliseconds for an HTTP/2 connection before it is gracefully \
          drained and replaced with a fresh one. When the timer expires the pool unregisters \
          from the Registry (so new requests go to a fresh connection), finishes any in-flight \
          requests, then terminates normally — the supervisor restarts it with a new DNS lookup. \
          Useful for Kubernetes headless-service load balancing where DNS entries rotate. \
          Defaults to `:infinity` (no age limit).
          """,
          default: :infinity
        ],
        max_connection_age_jitter: [
          type: :non_neg_integer,
          doc: """
          Random jitter in milliseconds added to `:max_connection_age`. Prevents multiple \
          pool shards from draining simultaneously (thundering-herd). The actual age used is \
          `max_connection_age + :rand.uniform(max_connection_age_jitter)`. \
          Defaults to `0` (no jitter).
          """,
          default: 0
        ]
      ]
    ]
  ]

  @typedoc """
  The `:name` provided to Finch in `start_link/1`.
  """
  @type name() :: atom()

  @type scheme() :: :http | :https

  @type scheme_host_port() :: {scheme(), host :: String.t(), port :: :inet.port_number()}

  @type pool_identifier() :: url :: String.t() | scheme_host_port() | Finch.Pool.t()

  @typedoc """
  Pool metrics returned by `get_pool_status/2` for a single pool.
  """
  @type pool_metrics() ::
          [Finch.HTTP1.PoolMetrics.t()]
          | [Finch.HTTP2.PoolMetrics.t()]

  @typedoc """
  Pool metrics grouped by pool identifier when querying the `:default` configuration.
  """
  @type default_pool_metrics() :: %{required(Finch.Pool.t()) => pool_metrics()}

  @type request_opt() ::
          {:pool_timeout, timeout()}
          | {:receive_timeout, timeout()}
          | {:request_timeout, timeout()}
          | {:pool_strategy, pool_strategy()}

  @typedoc """
  Pool strategy to choose a pool from a list of pools.

    * `{module, state}` - a module implementing `Finch.Pool.Strategy`
        (e.g. `{Finch.Pool.Strategy.RoundRobin, counter}`)
    * `{&module.select/2, state}` - same as above but avoids dynamic dispatch;
        use for performance-critical paths
    * `module` - a module implementing `Finch.Pool.Strategy`
        (e.g. `Finch.Pool.Strategy.Random`) that needs no state: `nil` will be passed as a default
    * a 1-arity function `fn entries -> chosen end` where `entries` is `nonempty_list(term())`
  """

  @type pool_strategy_state() :: term()
  @type pool_strategy_module() :: module()
  @type pool_strategy_module_with_state() :: {module(), pool_strategy_state()}
  @type pool_strategy_fun() :: (nonempty_list(term()) -> term())
  @type pool_strategy_fun_with_state() ::
          {(nonempty_list(term()), pool_strategy_state() -> term()), pool_strategy_state()}

  @type pool_strategy() ::
          pool_strategy_fun()
          | pool_strategy_fun_with_state()
          | pool_strategy_module_with_state()
          | pool_strategy_module()

  @typedoc """
  Options used by request functions.
  """
  @type request_opts() :: [request_opt()]

  @typedoc """
  Errors returned by Finch request functions.
  """
  @type error() :: Finch.Error.t() | Finch.HTTPError.t() | Finch.TransportError.t()

  @doc """
  Returns true if the term is any Finch error struct (`Finch.error()`).
  """
  defguard is_finch_error(term)
           when is_struct(term, Finch.Error) or is_struct(term, Finch.HTTPError) or
                  is_struct(term, Finch.TransportError)

  @typedoc """
  The reference used to identify a request sent using `async_request/3`.

  Use the `is_request_ref/1` guard when matching on async response messages in
  `c:GenServer.handle_info/2` or similar callbacks to ensure your code keeps
  working if the internal structure of the reference changes.
  """
  @opaque request_ref() :: Finch.Pool.Manager.request_ref()

  @doc """
  A guard that returns true if `ref` is a valid request reference from `async_request/3`.

  Use this guard when matching on async response messages in `c:GenServer.handle_info/2`
  so your code remains valid if the internal structure of the reference changes.

  ## Example

      require Finch

      def handle_info({ref, response}, state) when Finch.is_request_ref(ref) do
        # handle async response from Finch.async_request/3
      end

  """
  defguard is_request_ref(ref) when Finch.Pool.Manager.is_request_ref(ref)

  @typedoc """
  The stream function given to `stream/5`.
  """
  @type stream(acc) ::
          ({:status, integer}
           | {:headers, Mint.Types.headers()}
           | {:data, binary}
           | {:trailers, Mint.Types.headers()},
           acc ->
             acc)

  @typedoc """
  The stream function given to `stream_while/5`.
  """
  @type stream_while(acc) ::
          ({:status, integer}
           | {:headers, Mint.Types.headers()}
           | {:data, binary}
           | {:trailers, Mint.Types.headers()},
           acc ->
             {:cont, acc} | {:halt, acc})

  @typedoc """
  The request body function used with `{:stream, req_body_fun}` in `build/5`.
  """
  @type req_body_fun(acc) ::
          (acc -> {:data, binary(), acc} | {:done, acc} | {:halt, acc})

  @doc """
  Start an instance of Finch.

  ## Options

    * `:name` - The name of your Finch instance. Required.
    * `:pools` - A map of pool identifiers to configuration options. See the ":pools" subsection below.

  ### :name

  The name of your Finch instance. It is used to identify the instance when making requests
  and when calling other functions like `Finch.start_pool/3` or `Finch.get_pool_status/2`.

  #### Examples

      Finch.start_link(name: MyFinch)

  ### :pools

  A map where each key identifies a pool and each value is a keyword list of pool configuration
  options (see "[Pool Configuration Options](#start_link/1-pool-configuration-options)" below).
  Default is `%{default: [size: #{@default_pool_size}, count: #{@default_pool_count}]}`.

  **Pool keys** may be:

  * URL string – A binary URL. Pools created from URLs use the `:default` tag unless you
    use a `t:Finch.Pool.t/0` struct as the key instead.
  * `t:Finch.Pool.t/0` struct – Created with `Finch.Pool.new/2`. Use this when you need
    tagged pools (e.g. to run multiple pools for the same host with different configs).
  * URL string with `http+unix://` or `https+unix://` – For Unix domain sockets (e.g.
    `"http+unix:///tmp/socket"`)
  * `:default` – Catch-all. Any request whose pool is not in the map will use this config
    when its pool is started.

  When making a request with a `:pool_tag` option, that tag must exist in your pool configuration.
  If it does not, the request uses the `:default` configuration.

  #### Examples

      # URL keys (pool uses :default tag)
      Finch.start_link(
        name: MyFinch,
        pools: %{
          "https://api.example.com" => [size: 10, count: 2]
        }
      )

      # Tagged pools via Finch.Pool.new/2
      Finch.start_link(
        name: MyFinch,
        pools: %{
          Finch.Pool.new("https://api.example.com", tag: :bulk) => [size: 100, count: 1],
          Finch.Pool.new("https://api.example.com", tag: :realtime) => [size: 10, count: 2]
        }
      )

      # Unix socket
      Finch.start_link(
        name: MyFinch,
        pools: %{
          "http+unix:///tmp/socket" => [size: 5]
        }
      )

      # Custom default configuration
      Finch.start_link(
        name: MyFinch,
        pools: %{
          :default => [size: 25, count: 2]
        }
      )

  ### Multiple shards (`count` > 1) and `:pool_strategy`

  When `:count` is greater than 1, Finch starts that many shards for the same pool. Each request
  must pick one shard. **By default** Finch picks uniformly at random, which matches
  `Finch.Pool.Strategy.Random`.

  You can override selection per request with the `:pool_strategy` option (see `t:pool_strategy/0`
  and `Finch.Pool.Strategy`). Built-in modules include `Finch.Pool.Strategy.RoundRobin` and
  `Finch.Pool.Strategy.Hash` (stable mapping from a key to a shard, useful for affinity).
  You can also pass a custom module or function.

  ### Pool Configuration Options

  #{NimbleOptions.docs(@pool_config_schema)}
  """
  def start_link(opts) do
    name = finch_name!(opts)
    pools = Keyword.get(opts, :pools, []) |> pool_options!()
    {default_pool_config, pools} = Map.pop(pools, :default)

    config = %{
      registry_name: name,
      registry_listeners: Keyword.get(opts, :registry_listeners, []),
      supervisor_name: Pool.Manager.supervisor_name(name),
      supervisor_registry_name: Pool.Manager.supervisor_registry_name(name),
      default_pool_config: default_pool_config,
      pools: pools
    }

    Supervisor.start_link(__MODULE__, config, name: concat_name(name, "Supervisor"))
  end

  @doc """
  Finds a pool by its configuration and returns the pool pid.

  Returns `{:ok, pid}` if the pool exists, `:error` otherwise.

  This is useful for checking if a pool is available before making requests,
  or for advanced use cases where you need direct access to the pool process.

  ## Example

      case Finch.find_pool(MyFinch, Finch.Pool.new("https://api.internal", tag: :api)) do
        {:ok, pid} -> # Pool exists
        :error -> # Pool not found
      end
  """
  @spec find_pool(name(), Finch.Pool.t()) :: {:ok, pid()} | :error
  def find_pool(name, %Finch.Pool{} = pool) do
    case Pool.Manager.get_pool(name, pool, %{start_pool?: false}) do
      {pid, _mod} -> {:ok, pid}
      _ -> :error
    end
  end

  @doc """
  Starts a pool dynamically under Finch's internal supervision tree.

  Returns `:ok` if the pool was started or already exists.

  ## Options

  Same pool configuration options as `Finch.start_link/1`:
  `:size`, `:count`, `:protocols`, `:conn_opts`, etc.

  ## Example

      Finch.start_pool(MyFinch, Finch.Pool.new("https://api.example.com", tag: :api), size: 10)
  """
  @spec start_pool(name(), Finch.Pool.t(), keyword()) :: :ok
  def start_pool(name, pool, opts \\ [])

  def start_pool(name, %Finch.Pool{} = pool, opts) do
    # Avoid building the child_spec (cast_pool_opts, sanitize, etc.) if the pool already exists
    if Process.whereis(name) do
      supervisor_registry_name = Pool.Manager.supervisor_registry_name(name)

      case Pool.Manager.get_pool(supervisor_registry_name, pool, %{start_pool?: false}) do
        :not_found ->
          spec = Finch.Pool.child_spec([finch: name, pool: pool] ++ opts)
          supervisor_name = Pool.Manager.supervisor_name(name)

          case DynamicSupervisor.start_child(supervisor_name, spec) do
            {:ok, _pid} -> :ok
            {:error, {:already_started, _pid}} -> :ok
          end

        _ ->
          :ok
      end
    else
      raise ArgumentError, "Finch instance #{inspect(name)} is not running"
    end
  end

  def child_spec(opts) do
    %{
      id: finch_name!(opts),
      start: {__MODULE__, :start_link, [opts]}
    }
  end

  @impl true
  def init(config) do
    Finch.PoolMetrics.new(config.registry_name)

    children = [
      {Registry,
       keys: :duplicate,
       name: config.registry_name,
       listeners: config.registry_listeners,
       meta: [config: config]},
      {Registry, keys: :unique, name: config.supervisor_registry_name},
      {DynamicSupervisor, name: config.supervisor_name, strategy: :one_for_one},
      {Pool.Manager, config}
    ]

    Supervisor.init(children, strategy: :one_for_all)
  end

  defp finch_name!(opts) do
    Keyword.get(opts, :name) || raise(ArgumentError, "must supply a name")
  end

  defp pool_options!(pools) do
    {:ok, default} = NimbleOptions.validate([], @pool_config_schema)

    Enum.reduce(pools, %{default: valid_opts_to_map(default)}, fn {destination, opts}, acc ->
      with {:ok, valid_destination} <- cast_destination(destination),
           {:ok, valid_pool_opts} <- cast_pool_opts(opts) do
        Map.put(acc, valid_destination, valid_pool_opts)
      else
        {:error, reason} ->
          raise reason
      end
    end)
  end

  defp cast_destination(destination) do
    case destination do
      :default ->
        {:ok, destination}

      %Finch.Pool{} = pool ->
        {:ok, pool}

      {scheme, {:local, path}} when is_atom(scheme) and is_binary(path) ->
        ## TODO: Remove in Finch v1.0
        IO.warn("""
        Using {scheme, {:local, path}} as a pool key is deprecated. Use "#{scheme}+unix://#{path}" instead.

        For example:

            pools: %{
              "#{scheme}+unix://#{path}" => ...
            }
        """)

        {:ok, Finch.Pool.new({scheme, {:local, path}})}

      url when is_binary(url) ->
        {:ok, Finch.Pool.new(url)}

      _ ->
        {:error, %ArgumentError{message: "invalid destination: #{inspect(destination)}"}}
    end
  end

  @doc false
  def cast_pool_opts(opts) do
    with {:ok, valid} <- NimbleOptions.validate(opts, @pool_config_schema) do
      {:ok, valid_opts_to_map(valid)}
    end
  end

  defp valid_opts_to_map(valid) do
    # We need to enable keepalive and set the nodelay flag to true by default.
    transport_opts =
      valid
      |> get_in([:conn_opts, :transport_opts])
      |> List.wrap()
      |> Keyword.put_new(:timeout, @default_connect_timeout)
      |> Keyword.put_new(:nodelay, true)
      |> Keyword.put(:keepalive, true)

    conn_opts = valid[:conn_opts] |> List.wrap()

    # Only relevant to HTTP2, but just gracefully ignored in HTTP1.
    # Since we cannot handle server push responses, we need to disable the feature.
    client_settings =
      conn_opts
      |> Keyword.get(:client_settings, [])
      |> Keyword.put(:enable_push, false)

    ssl_key_log_file =
      Keyword.get(conn_opts, :ssl_key_log_file) || System.get_env("SSLKEYLOGFILE")

    ssl_key_log_file_device = ssl_key_log_file && File.open!(ssl_key_log_file, [:append])

    conn_opts =
      conn_opts
      |> Keyword.put(:ssl_key_log_file_device, ssl_key_log_file_device)
      |> Keyword.put(:transport_opts, transport_opts)
      |> Keyword.put(:protocols, valid[:protocols])
      |> Keyword.put(:client_settings, client_settings)

    mod =
      if :http1 in valid[:protocols] do
        Finch.HTTP1.Pool
      else
        Finch.HTTP2.Pool
      end

    %{
      mod: mod,
      size: valid[:size],
      count: valid[:count],
      conn_opts: conn_opts,
      conn_max_idle_time: to_native(valid[:conn_max_idle_time]),
      pool_max_idle_time: valid[:pool_max_idle_time],
      start_pool_metrics?: valid[:start_pool_metrics?],
      wait_for_server_settings?: valid[:http2][:wait_for_server_settings?],
      ping_interval: valid[:http2][:ping_interval],
      max_connection_age: valid[:http2][:max_connection_age],
      max_connection_age_jitter: valid[:http2][:max_connection_age_jitter]
    }
  end

  defp to_native(:infinity), do: :infinity
  defp to_native(time), do: System.convert_time_unit(time, :millisecond, :native)

  defp concat_name(name, suffix), do: :"#{name}.#{suffix}"

  defmacrop request_span(request, name, do: block) do
    quote do
      start_meta = %{request: unquote(request), name: unquote(name)}

      Finch.Telemetry.span(:request, start_meta, fn ->
        result = unquote(block)
        end_meta = Map.put(start_meta, :result, result)
        {result, end_meta}
      end)
    end
  end

  @doc """
  Builds an HTTP request to be sent with `request/3`, `async_request/3`, `stream/5`,
  or `stream_while/5`.

  Request `body` can be one of:

    * `nil` - no body is sent with the request.

    * `iodata` - the body to send for the request.

    * `{:stream, enumerable}` - stream request body chunks emitted by an `Enumerable`.

    * `{:stream, req_body_fun}` - stream request body chunks emitted by `req_body_fun`.
      Can only be used with `Finch.stream_while/5` on HTTP/1 pools.

      See `Finch.stream_while/5` for more information.

  ## Options

    * `:unix_socket` - Path to a Unix domain socket to connect to instead of the URL
      host/port. The URL scheme still determines whether HTTP or HTTPS is used.

    * `:pool_tag` - The tag to use when selecting which pool to use for this request.
      Defaults to `:default`.
  """
  @spec build(
          Request.method(),
          Request.url(),
          Request.headers(),
          Request.body(),
          Request.build_opts()
        ) ::
          Request.t()
  defdelegate build(method, url, headers \\ [], body \\ nil, opts \\ []), to: Request

  @doc """
  Streams an HTTP request and returns the accumulator.

  `resp_fun` receives a response entry and the accumulator `acc`, and must return
  the updated accumulator.

  Response entries are:

    * `{:status, status}` - the HTTP response status

    * `{:headers, headers}` - the HTTP response headers

    * `{:data, data}` - the HTTP response body chunk

    * `{:trailers, trailers}` - the HTTP response trailers

  See also `request/3`, `stream_while/5`.

  > ### HTTP2 streaming and back-pressure {: .warning}
  >
  > At the moment, streaming over HTTP2 connections do not provide
  > any back-pressure mechanism: this means the response will be
  > sent to the client as quickly as possible. Therefore, you must
  > not use streaming over HTTP2 for non-terminating responses or
  > when streaming large responses which you do not intend to keep
  > in memory.

  > ### Connection draining {: .info}
  >
  > If the HTTP/2 pool this request is dispatched to is currently draining (see
  > `http2: [max_connection_age: ...]`), the request is automatically retried on a fresh
  > pool. The retry is transparent to the caller. See `async_request/3` for the async
  > variant, which does not retry automatically.

  ## Options

  Shares options with `request/3`.

  ## Examples

      path = "/tmp/archive.zip"
      file = File.open!(path, [:write, :exclusive])
      url = "https://example.com/archive.zip"
      request = Finch.build(:get, url)

      Finch.stream(request, MyFinch, nil, fn
        {:status, status}, _acc ->
          IO.inspect(status)

        {:headers, headers}, _acc ->
          IO.inspect(headers)

        {:data, data}, _acc ->
          IO.binwrite(file, data)
      end)

      File.close(file)
  """
  @spec stream(Request.t(), name(), acc, stream(acc), request_opts()) ::
          {:ok, acc} | {:error, Finch.error(), acc}
        when acc: term()
  def stream(%Request{} = req, name, acc, fun, opts \\ []) when is_function(fun, 2) do
    validate_no_req_body_fun!(req, "Finch.stream/5")

    fun = fn entry, acc ->
      {:cont, fun.(entry, acc)}
    end

    stream_while(req, name, acc, fun, opts)
  end

  @doc """
  Streams an HTTP request until it finishes or is cancelled.

  ## Request body streaming

  When the request body is set to `{:stream, req_body_fun}` (see `build/5`), `req_body_fun`
  receives the accumulator `acc` and must return one of:

    * `{:data, chunk, acc}` - emit request body `chunk` and continue streaming

    * `{:done, acc}` - request body is done, `acc` is passed to `resp_fun`

    * `{:halt, acc}` - cancel the request and close the connection

  `{:stream, req_body_fun}` is currently only supported on HTTP/1 pools.

  ## Response streaming

  `resp_fun` receives a response entry and the accumulator `acc`, and must return one of:

    * `{:cont, acc}` - continue streaming

    * `{:halt, acc}` - cancel the request. On HTTP/1, this also closes the connection

  Response entries are:

    * `{:status, status}` - the HTTP response status

    * `{:headers, headers}` - the HTTP response headers

    * `{:data, data}` - the HTTP response body chunk

    * `{:trailers, trailers}` - the HTTP response trailers

  See also `request/3`, `stream/5`.

  > ### HTTP2 streaming and back-pressure {: .warning}
  >
  > At the moment, streaming over HTTP2 connections do not provide
  > any back-pressure mechanism: this means the response will be
  > sent to the client as quickly as possible. Therefore, you must
  > not use streaming over HTTP2 for non-terminating responses or
  > when streaming large responses which you do not intend to keep
  > in memory.

  > ### Connection draining {: .info}
  >
  > If the HTTP/2 pool this request is dispatched to is currently draining (see
  > `http2: [max_connection_age: ...]`), the request is automatically retried on a fresh
  > pool. The retry is transparent to the caller. See `async_request/3` for the async
  > variant, which does not retry automatically.

  ## Options

  Shares options with `request/3`.

  ## Examples

      path = "/tmp/archive.zip"
      file = File.open!(path, [:write, :exclusive])
      request = Finch.build(:get, "https://example.com/archive.zip")

      Finch.stream_while(request, MyFinch, nil, fn
        {:status, status}, acc ->
          IO.inspect(status)
          {:cont, acc}

        {:headers, headers}, acc ->
          IO.inspect(headers)
          {:cont, acc}

        {:data, data}, acc ->
          IO.binwrite(file, data)
          {:cont, acc}
      end)

      File.close(file)

  Uploading a file using `req_body_fun`:

      file = File.open!("/tmp/archive.zip", [:read])

      req_body_fun = fn file ->
        case IO.binread(file, 4096) do
          :eof -> {:done, file}
          data -> {:data, data, file}
        end
      end

      request = Finch.build(:post, "https://example.com/upload", [], {:stream, req_body_fun})

      resp_fun = fn
        {:status, status}, acc ->
          IO.inspect(status)
          {:cont, acc}

        {:headers, headers}, acc ->
          IO.inspect(headers)
          {:cont, acc}

        {:data, data}, acc ->
          IO.inspect(data)
          {:cont, acc}
      end

      {:ok, file} = Finch.stream_while(request, MyFinch, file, resp_fun)
      File.close(file)
  """
  @spec stream_while(Request.t(), name(), acc, stream_while(acc), request_opts()) ::
          {:ok, acc} | {:error, Finch.error(), acc}
        when acc: term()
  def stream_while(%Request{} = req, name, acc, fun, opts \\ []) when is_function(fun, 2) do
    request_span req, name do
      __stream__(req, name, acc, fun, opts)
    end
  end

  defp __stream__(req, name, acc, fun, opts, retries \\ 3)

  defp __stream__(%Request{} = req, name, acc, fun, opts, retries) do
    case get_pool(req, name, opts) do
      {pool, pool_mod} ->
        case pool_mod.request(pool, req, acc, fun, name, opts) do
          {:error, %Finch.Error{reason: :read_only}, _acc} when retries > 0 ->
            __stream__(req, name, acc, fun, opts, retries - 1)

          other ->
            other
        end

      _ ->
        {:error, Finch.Error.exception(:pool_not_available), acc}
    end
  end

  @doc """
  Sends an HTTP request and returns a `Finch.Response` struct.

  It can still raise exceptions if it was not possible to check out a connection in the given `:pool_timeout`.

  See also `stream/5`.

  > ### Connection draining {: .info}
  >
  > If the HTTP/2 pool this request is dispatched to is currently draining (see
  > `http2: [max_connection_age: ...]`), the request is automatically retried on a fresh
  > pool. The retry is transparent to the caller. See `async_request/3` for the async
  > variant, which does not retry automatically.

  ## Options

    * `:pool_timeout` - This timeout is applied when we check out a connection from the pool.
      Default value is `5_000`.

    * `:receive_timeout` - The maximum time to wait for each chunk to be received before returning an error.
      Default value is `15_000`.

    * `:request_timeout` - The amount of time to wait for a complete response before returning an error.
      This timeout only applies to HTTP/1, and its current implementation is a best effort timeout,
      it does not guarantee the call will return precisely when the time has elapsed.
      Default value is `:infinity`.

    * `:pool_strategy` - When the pool has multiple shards (`count: N`), selects which shards handles
      the request. Default is random selection. See `t:pool_strategy/0` for details.
  """
  @spec request(Request.t(), name(), request_opts()) ::
          {:ok, Response.t()}
          | {:error, Finch.error()}
  def request(req, name, opts \\ [])

  def request(%Request{} = req, name, opts) do
    validate_no_req_body_fun!(req, "Finch.request/3")
    Keyword.validate!(opts, [:pool_timeout, :receive_timeout, :request_timeout, :pool_strategy])

    request_span req, name do
      acc = {nil, [], [], []}

      fun = fn
        {:status, value}, {_, headers, body, trailers} ->
          {:cont, {value, headers, body, trailers}}

        {:headers, value}, {status, headers, body, trailers} ->
          {:cont, {status, headers ++ value, body, trailers}}

        {:data, value}, {status, headers, body, trailers} ->
          {:cont, {status, headers, [body | value], trailers}}

        {:trailers, value}, {status, headers, body, trailers} ->
          {:cont, {status, headers, body, trailers ++ value}}
      end

      case __stream__(req, name, acc, fun, opts) do
        {:ok, {status, headers, body, trailers}} ->
          {:ok,
           %Response{
             status: status,
             headers: headers,
             body: IO.iodata_to_binary(body),
             trailers: trailers
           }}

        {:error, error, _acc} ->
          {:error, error}
      end
    end
  end

  @doc """
  Sends an HTTP request and returns a `Finch.Response` struct
  or raises an exception in case of failure.

  See `request/3` for more detailed information.
  """
  @spec request!(Request.t(), name(), request_opts()) ::
          Response.t()
  def request!(%Request{} = req, name, opts \\ []) do
    case request(req, name, opts) do
      {:ok, resp} -> resp
      {:error, exception} -> raise exception
    end
  end

  @doc """
  Sends an HTTP request asynchronously, returning a request reference.

  If the request is sent using HTTP1, an extra process is spawned to
  consume messages from the underlying socket. The messages are sent
  to the current process as soon as they arrive, as a firehose.  If
  you wish to maximize request rate or have more control over how
  messages are streamed, a strategy using `request/3` or `stream/5`
  should be used instead.

  ## Receiving the response

  Response information is sent to the calling process as it is received
  in `{ref, response}` tuples.

  If the calling process exits before the request has completed, the
  request will be canceled.

  Responses include:

    * `{:status, status}` - HTTP response status
    * `{:headers, headers}` - HTTP response headers
    * `{:data, data}` - section of the HTTP response body
    * `{:error, exception}` - an error occurred during the request
    * `:done` - request has completed successfully

  On a successful request, a single `:status` message will be followed
  by a single `:headers` message, after which more than one `:data`
  messages may be sent. If trailing headers are present, a final
  `:headers` message may be sent. Any `:done` or `:error` message
  indicates that the request has succeeded or failed and no further
  messages are expected.

  ## Example

      iex> req = Finch.build(:get, "https://httpbin.org/stream/5")
      iex> ref = Finch.async_request(req, MyFinch)
      iex> flush()
      {ref, {:status, 200}}
      {ref, {:headers, [...]}}
      {ref, {:data, "..."}}
      {ref, :done}

  > ### Connection draining {: .info}
  >
  > Unlike `request/3` and `stream/5`, async requests are not automatically retried when a
  > pool is draining (see `http2: [max_connection_age: ...]`). If the caller receives
  > `{ref, {:error, %Finch.Error{reason: :read_only}}}`, it should retry by calling
  > `async_request/3` again.

  ## Options

  Shares options with `request/3`.
  """
  @spec async_request(Request.t(), name(), request_opts()) :: request_ref()
  def async_request(%Request{} = req, name, opts \\ []) do
    validate_no_req_body_fun!(req, "Finch.async_request/3")

    case get_pool(req, name, opts) do
      {pool, pool_mod} -> pool_mod.async_request(pool, req, name, opts)
      _ -> raise Finch.Error, :pool_not_available
    end
  end

  @doc """
  Cancels a request sent with `async_request/3`.
  """
  @spec cancel_async_request(request_ref()) :: :ok
  def cancel_async_request(request_ref) when Finch.Pool.Manager.is_request_ref(request_ref) do
    {pool_mod, _cancel_ref} = request_ref
    pool_mod.cancel_async_request(request_ref)
  end

  defp validate_no_req_body_fun!(%Request{body: {:stream, fun}}, caller)
       when is_function(fun, 1) do
    raise ArgumentError,
          "streaming request body with an accumulator function is not supported in #{caller}, " <>
            "use Finch.stream_while/5 instead"
  end

  defp validate_no_req_body_fun!(_req, _caller), do: :ok

  defp get_pool(%Request{scheme: scheme, unix_socket: unix_socket, pool_tag: tag}, name, opts)
       when is_binary(unix_socket) do
    pool = Finch.Pool.from_name({scheme, {:local, unix_socket}, 0, tag})
    Pool.Manager.get_pool(name, pool, opts)
  end

  defp get_pool(%Request{scheme: scheme, host: host, port: port, pool_tag: tag}, name, opts) do
    pool = Finch.Pool.from_name({scheme, host, port, tag})
    Pool.Manager.get_pool(name, pool, opts)
  end

  @doc """
  Get pool metrics.

  When given a URL or pool identifier tuple, this returns the metrics list for that specific
  pool. The number of items in the metrics list depends on the configured
  `:count` option and each entry will have a `pool_index` going from 1 to
  `:count`.

  When `:default` is provided, Finch returns the metrics for all pools started
  from the `:default` configuration. In this case the return value is a map
  keyed by each pool's `{scheme, host, port}` tuple with the corresponding
  metrics list as the value.

  The metrics struct depends on the pool scheme defined in the `:protocols`
  option: `Finch.HTTP1.PoolMetrics` for `:http1` and `Finch.HTTP2.PoolMetrics`
  for `:http2`. See the documentation for those modules for more details.

  `{:error, :not_found}` is returned in the following scenarios:

    * There is no pool registered for the given Finch instance and pool identifier.
    * The pool has `start_pool_metrics?: false` (the default).
    * `:default` is provided but no pools have been started from the
      `:default` configuration (or none have metrics enabled).

  ## Examples

      iex> Finch.get_pool_status(MyFinch, "https://httpbin.org")
      {:ok, [
        %Finch.HTTP1.PoolMetrics{
          pool_index: 1,
          pool_size: 50,
          available_connections: 43,
          in_use_connections: 7
        },
        %Finch.HTTP1.PoolMetrics{
          pool_index: 2,
          pool_size: 50,
          available_connections: 37,
          in_use_connections: 13
        }]
      }

      iex> Finch.get_pool_status(MyFinch, :default)
      {:ok,
       %{
         %Finch.Pool{host: "httpbin.com", port: 443, scheme: :https, tag: :default} => [
           %Finch.HTTP1.PoolMetrics{
             pool_index: 1,
             pool_size: 50,
             available_connections: 43,
             in_use_connections: 7
           }
         ]
       }}
  """
  @spec get_pool_status(name(), :default | pool_identifier()) ::
          {:ok, pool_metrics()}
          | {:ok, default_pool_metrics()}
          | {:error, :not_found}

  def get_pool_status(finch_name, :default) do
    finch_name
    |> Pool.Manager.get_default_pools()
    |> Enum.reduce(%{}, fn {_sup_pid, {pool_name, pool_mod}}, acc ->
      case pool_mod.get_pool_status(finch_name, pool_name) do
        {:ok, metrics} ->
          pool_id = Pool.from_name(pool_name)
          Map.put(acc, pool_id, metrics)

        {:error, :not_found} ->
          acc
      end
    end)
    |> case do
      result when result == %{} -> {:error, :not_found}
      result -> {:ok, result}
    end
  end

  def get_pool_status(finch_name, pool_identifier) do
    pool = resolve_pool(pool_identifier)

    case Pool.Manager.get_pool_mod(finch_name, pool) do
      {pool_name, pool_mod} ->
        pool_mod.get_pool_status(finch_name, pool_name)

      :not_found ->
        {:error, :not_found}
    end
  end

  @doc """
  Sends an HTTP/2 PING frame and waits for PONG.

  Returns `{:ok, rtt_ms}` where `rtt_ms` is the round-trip time in native time units,
  or `{:error, reason}` if the ping fails.

  This is only supported for HTTP/2 pools. Returns `{:error, :not_http2}` for
  HTTP/1 pools.

  ## Examples

      {:ok, rtt} = Finch.ping(MyFinch, "https://example.com")
      IO.puts("RTT: \#{rtt}ms")

  """
  @spec ping(name(), pool_identifier()) :: {:ok, integer()} | {:error, term()}
  def ping(name, pool_identifier) do
    pool = resolve_pool(pool_identifier)

    case Pool.Manager.get_pool(name, pool) do
      {pid, Finch.HTTP2.Pool} ->
        Finch.HTTP2.Pool.ping(pid)

      {_pid, _mod} ->
        {:error, :not_http2}

      :not_found ->
        {:error, :not_found}
    end
  end

  @doc """
  Stops the pool of processes associated with the given pool identifier.

  This function can be invoked to manually stop the pool for the given identifier
  when you know it's not going to be used anymore.

  Note that this function is not safe with respect to concurrent requests. Invoking it while
  another request to the same pool is taking place might result in the failure of that request.
  It is the responsibility of the client to ensure that no request to the same pool is taking
  place while this function is being invoked.
  """
  @spec stop_pool(name(), pool_identifier()) :: :ok | {:error, :not_found}
  def stop_pool(finch_name, pool_identifier) do
    pool = resolve_pool(pool_identifier)

    case Pool.Manager.get_pool_supervisor(finch_name, pool) do
      :not_found ->
        {:error, :not_found}

      {pid, pool_name, _pool_mod, _pool_count, _pool_config} ->
        result = Supervisor.stop(pid)
        Finch.PoolMetrics.delete_pool(finch_name, pool_name)
        result
    end
  end

  @doc """
  Returns the current worker count for the given pool.

  Returns `{:ok, count}` if the pool exists, `{:error, :not_found}` otherwise.

  ## Examples

      {:ok, count} = Finch.get_pool_count(MyFinch, "https://example.com")
  """
  @spec get_pool_count(name(), pool_identifier()) :: {:ok, pos_integer()} | {:error, :not_found}
  def get_pool_count(finch_name, pool_identifier) do
    pool = resolve_pool(pool_identifier)

    case Pool.Manager.get_pool_supervisor(finch_name, pool) do
      :not_found -> {:error, :not_found}
      {_pid, _pool_name, _pool_mod, pool_count, _pool_config} -> {:ok, pool_count}
    end
  end

  @doc """
  Dynamically changes the number of pool workers for the given pool.

  Returns `:ok` on success, `{:error, :not_found}` if the pool doesn't exist.

  Works with all kinds of pools, but note that `:default` pools must have
  been materialized by at least one request before they can be resized.

  ## Examples

      :ok = Finch.set_pool_count(MyFinch, "https://example.com", 4)
  """
  @spec set_pool_count(name(), pool_identifier(), pos_integer()) :: :ok | {:error, term()}
  def set_pool_count(finch_name, pool_identifier, count) when is_integer(count) and count > 0 do
    pool = resolve_pool(pool_identifier)
    Pool.Manager.set_pool_count(finch_name, pool, count)
  end

  defp resolve_pool(%Finch.Pool{} = pool), do: pool

  defp resolve_pool({scheme, host, port}),
    do: Finch.Pool.from_name({scheme, host, port, :default})

  defp resolve_pool(url) when is_binary(url), do: Finch.Pool.new(url)
end