conn_pool

defmodule Conn.Pool do
  use Agent

  require Logger

  @moduledoc """
  Connection pool helps storing, sharing and using connections. It also make its
  possible to use the same connection concurrently. For example, if there exists
  remote API accessible via websocket, pool can provide shared access, making
  queues of calls. If connection is closed/expires — pool will reinitialize or
  drop it and awaiting calls will be moved to another connection queue.

  Start pool via `start_link/1` or `start/1`. Add connections via `init/3` or
  `Conn.init/2` → `put!/2`. Make calls from pool via `call/4`.

  In the following examples, `%Conn.Agent{}` represents connection to some
  `Agent` that can be created separately or via `Conn.init/2`. Available methods
  of interaction with `Agent` are `:get`, `:get_and_update`, `:update` and
  `:stop` (see `Conn.methods!/1`). This type of connection means to exist only
  as an example for doctests. More meaningful example would be `Conn.Plug`
  wrapper. Also, see `Conn` docs for detailed example on `Conn` protocol
  implementation and use.

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
      iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.extra(pool, id, type: :agent) # add `extra` info
      {:ok, nil}
      #
      # Pool wraps conn into `%Conn{}` struct.
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> info.extra
      [type: :agent]
      iex> info.methods
      [:get, :get_and_update, :update, :stop]
      iex> ^agent = Conn.resource(info.conn)
      iex> Conn.Pool.call(pool, agent, :get, & &1)
      {:ok, 42}
      #
      # Now we use filter (& &1.extra[:type] != :agent).
      iex> Conn.Pool.call(pool, agent, :get, & &1.extra[:type] != :agent, & &1)
      {:error, :filter}
      #
      iex> Conn.Pool.call(pool, agent, :badmethod)
      {:error, :method}
      iex> Conn.Pool.call(pool, :badres, :badmethod)
      {:error, :resource}

  In the above example connection was initialized and used directly from pool.
  `%Conn{}.extra` information that was given via `Conn.Pool.extra/3` was used to
  filter conn to be selected in `Conn.Pool.call/5` call.

  In the following example connection will be added via `put/2`.

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
      iex> {:ok, conn} = Conn.init(%Conn.Agent{}, res: agent)
      iex> Conn.Pool.put!(pool, %Conn{conn: conn, revive: true})
      iex> Process.alive?(agent) && not Conn.Pool.empty?(pool, agent)
      true
      # There are conns for resource `agent` in this pool.
      iex> Conn.Pool.call(pool, agent, :stop)
      :ok
      iex> not Process.alive?(agent)
      true
      iex> :timer.sleep(10)
      iex> Conn.Pool.empty?(pool, agent)
      true
      # Now conn is `:closed` and will be reinitialized by pool,
      # but:
      iex> {:error, :dead, :infinity, conn} == Conn.init(conn)
      true
      # `Conn.init/2` suggests to never reinitialize again
      # (`:infinity` timeout) so pool will just drop this conn.
      iex> Conn.Pool.resources(pool)
      []

  Also, TTL value could be provided. By default expired connection is revived.

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
      iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.update(pool, agent, & %{&1 | ttl: 50})
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> info.ttl
      50
      iex> Conn.Pool.call(pool, agent, :get, & &1)
      {:ok, 42}
      iex> :timer.sleep(50)
      :ok
      #
      # Next call will fail because all the conns are expired and conn
      # will not be revived (%Conn{}.revive == false).
      #
      iex> Conn.Pool.call(pool, agent, :get, & &1)
      {:error, :resource}
      iex> Conn.Pool.resources(pool)
      []
      iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.call(pool, agent, :get, & &1+1)
      {:ok, 43}
      # That's because conn that is alive exists.
      #
      iex> Conn.Pool.update(pool, agent, & %{&1 | ttl: 10, revive: true})
      iex> :timer.sleep(10)
      iex> Conn.Pool.call(pool, agent, :get, & &1+1)
      {:error, :resource}

  ## Name registration

  An `Conn.Pool` is bound to the same name registration rules as `GenServer`s.
  Read more about it in the `GenServer` docs.
  """

  @type t :: GenServer.t()
  @type id :: pos_integer
  @type reason :: any
  @type filter :: (Conn.info() -> as_boolean(term()))

  # Generate uniq id.
  defp gen_id, do: now()

  # Current monotonic time.
  defp now(), do: System.monotonic_time()

  @doc """
  Starts pool as a linked process.
  """
  @spec start_link(GenServer.options()) :: GenServer.on_start()
  def start_link(opts \\ []), do: AgentMap.start_link(opts)

  @doc """
  Starts pool. See `start_link/2` for details.
  """
  @spec start(GenServer.options()) :: GenServer.on_start()
  def start(opts \\ []), do: AgentMap.start(opts)

  @doc """
  Initialize connection in a separate `Task` and use `put/2` to add connection
  that is wrapped in `%Conn{}`.

  By default, init time is limited by `5000` ms and pool will revive closed
  connection using `init_args` provided.

  ## Returns

    * `{:ok, id}` in case of `Conn.init/2` returns `{:ok, conn}`, where `id` is
      the identifier that can be used to refer `conn`;
    * `{:error, :timeout}` if `Conn.init/2` returns `{:ok, :timeout}`;
    * `{:error, reason}` in case `Conn.init/2` returns an arbitrary error.
  """
  @spec init(Conn.Pool.t(), Conn.t() | Conn.info(), any) ::
          {:ok, id} | {:error, :timeout | reason}

  def init(pool, info_or_conn, init_args \\ nil)

  def init(pool, %Conn{conn: conn} = info, init_args) do
    task =
      Task.async(fn ->
        Conn.init(conn, init_args)
      end)

    case Task.yield(task, :infinity) || Task.shutdown(task) do
      {:ok, {:ok, conn}} ->
        {:ok, put!(pool, %{info | conn: conn, init_args: init_args})}

      {:ok, err} ->
        err

      nil ->
        {:error, :timeout}
    end
  end

  def init(pool, conn, init_args) do
    init(pool, %Conn{conn: conn}, init_args)
  end

  # Add conn to pool.
  defp add(pool, %{last_init: :never} = info) do
    add(pool, %{info | last_init: now()})
  end

  defp add(pool, info) do
    id = gen_id()
    AgentMap.put(pool, {:conn, id}, info)

    r = Conn.resource(info.conn)

    AgentMap.update(pool, {:res, r}, fn
      nil -> [id]
      ids -> [id | ids]
    end)

    id
  end

  defp add_methods!(info) do
    if info.methods && is_list(info.methods) do
      info
    else
      case Conn.methods!(info.conn) do
        {methods, conn} when is_list(methods) ->
          %{info | methods: methods, conn: conn}

        methods when is_list(methods) ->
          %{info | methods: methods}

        ret ->
          raise("""
            Conn.methods!/1 call expected to return [method]
            or {[method], updated conn}, but instead #{inspect(ret)} returned.
          """)
      end
    end
  end

  @doc """
  Adds given `%Conn{}` to pool, returning id to refer it later.
  This call always refresh data cached in `%Conn{}.methods`.

  ## Example

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, conn} = Conn.init(%Conn.Agent{}, fn -> 42 end)
      iex> ttl = 100 # ms
      iex> info = %Conn{
      ...>   conn: conn,
      ...>   extra: :extra,
      ...>   ttl: ttl
      ...> }
      iex> id = Conn.Pool.put!(pool, info)
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> ^ttl = info.ttl
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> info.extra
      :extra
      #
      # let's make some tweak
      #
      iex> {:ok, info} = Conn.Pool.pop(pool, id)
      iex> id = Conn.Pool.put!(pool, %{info| ttl: :infinity})
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> info.ttl
      :infinity
  """
  @spec put!(Conn.Pool.t(), Conn.info()) :: id
  def put!(pool, info) do
    add(pool, add_methods!(info))
  end

  defp delete(pool, id) do
    case info(pool, id) do
      {:ok, info} ->
        AgentMap.delete(pool, {:conn, id})

        AgentMap.cast(
          pool,
          fn
            [[^id]] ->
              :drop

            [ids] ->
              [List.delete(ids, id)]

            [nil] ->
              :id
          end,
          [{:res, Conn.resource(info.conn)}]
        )

      :error ->
        :ignore
    end
  end

  @doc """
  Deletes connection and returns corresponding `%Conn{}` struct.

  Returns only after all calls awaiting execution on this conn are made.

  ## Returns

    * `{:ok, Conn struct}` in case conn with such id exists;
    * `:error` — otherwise.

  ## Examples

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
      iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.call(pool, agent, :get, & &1)
      {:ok, 42}
      iex> {:ok, %Conn{conn: conn}} = Conn.Pool.pop(pool, id)
      iex> Conn.Pool.pop(pool, id)
      :error
      iex> Conn.Pool.call(pool, agent, :get, & &1)
      {:error, :resource}
      # But, still:
      iex> Agent.get(Conn.resource(conn), & &1)
      42

  Also, you can pop only those connections that satisfy the specified filter.

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start_link(fn -> 42 end)
      iex> {:ok, id1} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> {:ok, id2} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> {:ok, id3} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.extra(pool, id1, :takeme)
      {:ok, nil}
      iex> Conn.Pool.extra(pool, id3, :takeme)
      iex> Conn.Pool.pop(pool, agent, & &1.extra == :takeme)
      iex> Conn.Pool.empty?(pool, agent)
      false
      iex> Conn.Pool.pop(pool, id2)
      iex> :timer.sleep(10)
      iex> Conn.Pool.empty?(pool, agent)
      true

  Also, see example for `put!/2`.
  """
  @spec pop(Conn.Pool.t(), Conn.id()) :: {:ok, Conn.info()} | :error
  def pop(pool, id) when is_integer(id) do
    AgentMap.get_and_update(pool, {:conn, id}, fn
      nil ->
        {:error}

      info ->
        delete(pool, id)
        {{:ok, info}, nil}
    end)
  end

  @doc """
  Pops conns to `resource` that satisfy `filter`.
  """
  @spec pop(Conn.Pool.t(), Conn.resource(), filter) :: [Conn.info()]
  def pop(pool, resource, filter) when is_function(filter, 1) do
    pool = AgentMap.new(pool)
    ids = pool[{:res, resource}] || []

    for id <- ids, filter.(pool[{:conn, id}]) do
      pop(pool, id)
    end
  end

  @doc """
  Applies given `fun` to every `resource` conn.
  """
  @spec map(Conn.Pool.t(), Conn.resource(), (Conn.info() -> a)) :: [a] when a: var
  def map(pool, resource, fun) when is_function(fun, 1) do
    pool = AgentMap.new(pool)
    ids = pool[{:res, resource}] || []

    for id <- ids, do: fun.(pool[{:conn, id}])
  end

  @doc """
  Update every conn to `resource` with `fun`.
  This function always returns `:ok`.

  ## Example

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start(fn -> 42 end)
      iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.call(pool, agent, :get, & &1.extra == :extra, & &1)
      {:error, :filter}
      #
      iex> Conn.Pool.update(pool, agent, & %{&1 | extra: :extra})
      :ok
      iex> Conn.Pool.call(pool, agent, :get, & &1.extra == :extra, & &1)
      {:ok, 42}
  """
  @spec update(Conn.Pool.t(), Conn.resource(), (Conn.info() -> Conn.info())) :: :ok
  def update(pool, resource, fun) when is_function(fun, 1) do
    update(pool, resource, fn _ -> true end, fun)
  end

  @doc """
  Updates connections to `resource` that satisfy given `filter`. This function
  always returns `:ok`.

  ## Example

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start(fn -> 42 end)
      iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> filter = & &1.extra == :extra
      iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
      {:error, :filter}
      #
      iex> Conn.Pool.update(pool, agent, & %{&1 | extra: :extra})
      :ok
      iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
      {:ok, 42}
      iex> Conn.Pool.update(pool, agent, filter, &Map.put(&1, :extra, nil))
      iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
      {:error, :filter}
  """
  @spec update(Conn.Pool.t(), Conn.resource(), filter, (Conn.info() -> Conn.info())) :: :ok
  def update(pool, resource, filter, fun) when is_function(fun, 1) and is_function(filter, 1) do
    pool = AgentMap.new(pool)
    ids = pool[{:res, resource}] || []

    self = self()

    for id <- ids do
      AgentMap.update(pool, {:conn, id}, fn info ->
        if filter.(info) do
          try do
            add_methods!(fun.(info))
          rescue
            e ->
              Process.exit(self, """
              Error while making Conn.methods!/1 call on #{inspect(info.conn)}:
              #{inspect(Exception.message(e))}.
              """)

              info
          end
        else
          info
        end
      end)
    end

    :ok
  end

  @doc """
  Is `pool` has conns to the given `resource`?
  """
  @spec empty?(Conn.Pool.t(), Conn.resource()) :: boolean
  def empty?(pool, resource) do
    not AgentMap.has_key?(pool, {:res, resource})
  end

  @doc """
  Returns all the known resources.
  """
  @spec resources(Conn.Pool.t()) :: [Conn.resource()]
  def resources(pool) do
    Enum.flat_map(AgentMap.keys(pool), fn
      {:res, r} -> [r]
      _ -> []
    end)
  end

  # Returns {:ok, ids} | {:error, :resources} | {:error, method}
  # | {:error, filter}.
  defp filter(pool, resource, method, filter, except: except) when is_function(filter, 1) do
    pool = AgentMap.new(pool)
    ids = pool[{:res, resource}] || []
    import Enum

    ids =
      for id <- ids -- except do
        {id, pool[{:conn, id}]}
      end

    with {_, [_ | _] = ids} <- {:r, filter(ids, fn {_, c} -> not c.closed end)},
         {_, [_ | _] = ids} <- {:m, filter(ids, fn {_, c} -> method in c.methods end)},
         {_, [_ | _] = ids} <- {:f, filter(ids, fn {_, c} -> filter.(c) end)} do
      {:ok, for({id, _} <- ids, do: id)}
    else
      {:r, []} -> {:error, :resource}
      {:m, []} -> {:error, :method}
      {:f, []} -> {:error, :filter}
    end
  end

  defp _revive(pool, id, info) do
    case Conn.init(info.conn, info.init_args) do
      {:ok, conn} ->
        res = Conn.resource(conn)

        AgentMap.cast(pool, {:res, res}, fn
          nil -> [id]
          ids -> [id | ids]
        end)

        %{
          info
          | conn: conn,
            closed: false,
            last_init: now(),
            last_call: :never,
            timeout: 0
        }

      {:error, reason, :infinity, conn} ->
        info = %{info | conn: conn}

        Logger.error("""
        Failed to reinitialize connection.
        Reason: #{inspect(reason)}.
        Connection info: #{inspect(info)}.
        Stop trying.
        """)

        delete(pool, id)
        nil

      {:error, reason, timeout, conn} ->
        info = %{info | conn: conn}

        Logger.warn("""
        Failed to reinitialize connection with id #{id}.
        Reason: #{inspect(reason)}.
        Connection info: #{inspect(info)}.
        Will try again in #{to_ms(timeout)} ms.
        """)

        Process.sleep(timeout)
        _revive(pool, id, info)
    end
  end

  defp revive(pool, id, info) do
    AgentMap.cast(pool, {:conn, id}, fn nil ->
      _revive(pool, id, info)
    end)
  end

  # Estimated time to wait until call can be made on this id.
  defp waiting_time(pool, id) do
    with {:ok, info} <- AgentMap.fetch(pool, {:conn, id}) do
      {sum, n} =
        Enum.reduce(info.stats, {0, 0}, fn {_key, {avg, num}}, {sum, n} ->
          {sum + avg * num, n + num}
        end)

      if n == 0 do
        {:ok, info.timeout}
      else
        {:ok, sum / n * AgentMap.queue_len(pool, {:conn, id}) + info.timeout}
      end
    end
  end

  # Select the best conn.
  defp select(pool, resource, method, filter, opts \\ [except: []]) do
    with {:ok, ids} <- filter(pool, resource, method, filter, opts) do
      {:ok, Enum.min_by(ids, &waiting_time(pool, &1))}
    end
  end

  defp update_stats(info, method, start) do
    stop = System.monotonic_time()

    update_in(info.stats[method], fn
      nil -> {stop - start, 1}
      {avg, num} -> {(avg * num + stop - start) / (num + 1), num + 1}
    end)
  end

  defp to_ms(native) do
    System.convert_time_unit(native, :native, :milliseconds)
  end

  defp to_native(ms) do
    System.convert_time_unit(ms, :milliseconds, :native)
  end

  defp expired?(%{ttl: :infinity}), do: false

  defp expired?(info) do
    info.last_init + to_native(info.ttl) < now()
  end

  defp merge(info, conn, t \\ 0) do
    %{info | conn: conn, last_call: now(), timeout: t}
  end

  defp close(info, pool, status \\ :ok) do
    {:conn, id} = Process.get(:"$key")

    delete(pool, id)

    if info.revive == :force || (status != :ok && info.revive) do
      revive(pool, id, info)
    end

    %{info | closed: true, timeout: :infinity}
  end

  defp _call(%{closed: true} = info, {pool, resource, method, filter, _payload} = args) do
    {:conn, id} = Process.get(:"$key")

    case select(pool, resource, method, filter, except: [id]) do
      {:ok, id} ->
        key = {:conn, id}
        fun = &_call(&1, args)
        # Make chain call while not changing conn.
        {:chain, {key, fun}, info}

      err ->
        # Return an error while not changing conn.
        {err}
    end
  end

  defp _call(info, {pool, resource, method, filter, payload} = args) do
    {:conn, id} = Process.get(:"$key")

    if expired?(info) do
      delete(pool, id)
      if info.revive, do: revive(pool, id, info)
      _call(%{info | closed: true}, args)
    else
      start = now()
      timeout = to_native(info.timeout)

      # Time to wait.
      last_call = info.last_call

      ttw = (last_call == :never && 0) || last_call + timeout - start

      if to_ms(ttw) < 50 do
        if ttw > 0 do
          Process.sleep(to_ms(ttw))
        end

        try do
          case Conn.call(info.conn, method, payload) do
            {:noreply, conn} ->
              {:ok, merge(info, conn) |> update_stats(method, start)}

            {:noreply, t, conn} when t in [:infinity, :closed] ->
              {:ok,
               info
               |> merge(conn, :infinity)
               |> update_stats(method, start)
               |> close(pool)}

            {:noreply, timeout, conn} ->
              {:ok,
               info
               |> merge(conn, timeout)
               |> update_stats(method, start)}

            {:reply, r, conn} ->
              {{:ok, r}, merge(info, conn) |> update_stats(method, start)}

            {:reply, r, t, conn} when t in [:infinity, :closed] ->
              {{:ok, r},
               info
               |> merge(conn, :infinity)
               |> update_stats(method, start)
               |> close(pool)}

            {:reply, r, timeout, conn} ->
              {{:ok, r},
               info
               |> merge(conn, timeout)
               |> update_stats(method, start)}

            {:error, :closed} ->
              delete(pool, id)
              if info.revive, do: revive(pool, id, info)
              _call(%{info | closed: true}, args)

            {:error, reason, conn} ->
              {{:error, reason}, merge(info, conn)}

            {:error, reason, t, conn} when t in [:infinity, :closed] ->
              {{:error, reason},
               info
               |> merge(conn, :infinity)
               |> close(pool, :error)}

            {:error, reason, timeout, conn} ->
              {{:error, reason}, merge(info, conn, timeout)}

            err ->
              Logger.warn("Conn.call returned unexpected: #{inspect(err)}.")
              {err}
          end
        rescue
          e ->
            if info.unsafe do
              raise e
            else
              {:error, e}
            end
        end
      else
        # ttw > 50 ms.
        with {:ok, id} <- select(pool, resource, method, filter, except: [id]),
             {:ok, potential_ttw} <- waiting_time(pool, id),
             true <- ttw > potential_ttw do
          key = {:conn, id}
          fun = &_call(&1, args)
          {:chain, {key, fun}, info}
        else
          _ ->
            Process.sleep(20)
            _call(info, args)
        end
      end
    end
  end

  @doc """
  Select one of the connections to given `resource` and make `Conn.call/3` via
  given `method` of interaction.

  Optional `filter` param could be provided in form of `(%Conn{} ->
  as_boolean(term()))` callback.

  Pool respects refresh timeout value returned by `Conn.call/3`. After each call
  `:timeout` field of the corresponding `%Conn{}` struct is rewrited.

  ## Returns

    * `{:error, :resource}` if there is no conns to given `resource`;
    * `{:error, :method}` if there exists conns to given `resource` but they does
    not provide given `method` of interaction;
    * `{:error, :filter}` if there is no conns satisfying `filter`;

    * `{:error, :timeout}` if `Conn.call/3` returned `{:error, :timeout, _}`
      and there is no other connection capable to make this call;
    * and `{:error, reason}` in case of `Conn.call/3` returned arbitrary error.

  In case of `Conn.call/3` returns `{:error, :timeout | reason, _}`, `Conn.Pool`
  will use time penalties series, defined per pool (see `start_link/2`) or per
  connection (see `%Conn{} :penalties` field).

    * `{:ok, reply} | :ok` in case of success.
  """
  @spec call(Conn.Pool.t(), Conn.resource(), Conn.method(), any) ::
          :ok | {:ok, Conn.reply()} | {:error, :resource | :method | :timeout | reason}
  @spec call(Conn.Pool.t(), Conn.resource(), Conn.method(), filter, any) ::
          :ok | {:ok, Conn.reply()} | {:error, :resource | :method | :filter | :timeout | reason}
  def call(pool, resource, method, payload \\ nil) do
    call(pool, resource, method, fn _ -> true end, payload)
  end

  def call(pool, resource, method, filter, payload) when is_function(filter, 1) do
    with {:ok, id} <- select(pool, resource, method, filter) do
      key = {:conn, id}
      fun = &_call(&1, {pool, resource, method, filter, payload})
      AgentMap.get_and_update(pool, key, fun)
    end
  end

  @doc """
  `:extra` field of `%Conn{}` is intended for filtering conns while making call.
  Calling `extra/3` will change `:extra` field of connection with given `id`,
  while returning the old value in form of `{:ok, old extra}` or `:error` if
  pool don't known conn with such id.

  ## Example

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, agent} = Agent.start(fn -> 42 end)
      iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, res: agent)
      iex> Conn.Pool.extra(pool, id, :extra)
      {:ok, nil}
      iex> Conn.Pool.extra(pool, id, :some)
      {:ok, :extra}
      iex> badid = -1
      iex> Conn.Pool.extra(pool, badid, :some)
      :error
      #
      iex> filter = & &1.extra == :extra
      iex> Conn.Pool.call(pool, agent, :get, filter, & &1)
      {:error, :filter}
      #
      # but:
      iex> Conn.Pool.call(pool, agent, :get, & &1.extra == :some, & &1)
      {:ok, 42}
  """
  @spec extra(Conn.Pool.t(), id, any) :: {:ok, any} | :error
  def extra(pool, id, extra) do
    with {:ok, info} <- info(pool, id) do
      AgentMap.put(pool, {:conn, id}, %{info | extra: extra})
      {:ok, info.extra}
    end
  end

  @doc """
  Retrives connection wraped in a `%Conn{}`.
  Returns `{:ok, Conn.info}` or `:error`.

  ## Example

      iex> {:ok, pool} = Conn.Pool.start_link()
      iex> {:ok, id} = Conn.Pool.init(pool, %Conn.Agent{}, fn -> 42 end)
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> info.extra
      nil
      iex> Conn.Pool.extra(pool, id, :extra)
      {:ok, nil}
      iex> {:ok, info} = Conn.Pool.info(pool, id)
      iex> info.extra
      :extra
  """
  @spec info(Conn.Pool.t(), id) :: {:ok, %Conn{}} | :error
  def info(pool, id) when is_integer(id) do
    AgentMap.fetch(pool, {:conn, id})
  end
end