podbox_ash/lib/podbox/download/request.ex

defmodule Podbox.Download.Request do
  @moduledoc """
  Handles a Mint HTTP request.
  """

  alias Podbox.{
    Download,
    Download.Asset,
    Download.BadRedirectError,
    Download.HttpError,
    Download.InvalidUriError,
    Download.Storage,
    Download.UnexpectedStatusError,
    Download.UnsupportedSchemeError,
    Podcast
  }

  @vsn :podbox |> Application.spec(:vsn) |> to_string()
  @max_req_time :timer.minutes(45)

  @doc "Perform an HTTP request"
  @spec request(Asset.t()) :: :ok | {:error, any}
  def request(asset) do
    Task.async(fn ->
      with {:ok, uri} <- validate_uri(asset.uri) do
        perform_request(asset, uri)
      end
    end)
    |> Task.await(@max_req_time)
  end

  defp perform_request(asset, uri) do
    with {:ok, headers} <- default_headers(asset, uri) do
      storage = Storage.init(asset)
      req = Finch.build(:get, uri, headers)
      ref = Finch.async_request(req, Download.Finch)

      storage =
        if Storage.status(storage) != :ok do
          :ok = Storage.destroy(storage)
          Storage.init(asset)
        else
          storage
        end

      state = %{
        asset: asset,
        method: :get,
        req_headers: headers,
        resp_headers: [],
        status: nil,
        storage: storage,
        transferred_bytes: 0,
        uri: uri
      }

      handle_response(ref, req, state)
    end
  end

  defp handle_response(ref, req, state) do
    receive do
      {^ref, {:status, status}} when status in [200, 201, 202, 203, 206] ->
        handle_response(ref, req, %{state | status: status})

      {^ref, {:status, status}} when status in [301, 302] ->
        handle_response(ref, req, %{state | status: status})

      {^ref, {:status, status}} ->
        opts =
          state
          |> Map.take([:asset, :method, :req_headers, :uri])
          |> Map.put(:status, status)
          |> Enum.to_list()

        {:error, UnexpectedStatusError.exception(opts)}

      {^ref, {:headers, headers}} when state.status in [301, 302] ->
        with {:ok, location} <- fetch_location_header(headers, state),
             :ok <- Finch.cancel_async_request(ref),
             {:ok, uri} <- URI.new(location) do
          perform_request(state.asset, uri)
        end

      {^ref, {:headers, headers}} ->
        handle_response(ref, req, %{state | resp_headers: headers})

      {^ref, {:data, data}} when state.asset.state == :dequeued ->
        transferred_bytes = byte_size(data)

        params = %{
          total_bytes: get_header(state.resp_headers, "content-length"),
          content_type: get_header(state.resp_headers, "content-type"),
          transferred_bytes: transferred_bytes
        }

        asset = Download.started!(state.asset, params)
        {:ok, storage} = Storage.prepare(state.storage)
        {:ok, storage} = Storage.append(storage, data)

        handle_response(ref, req, %{
          state
          | asset: asset,
            transferred_bytes: transferred_bytes,
            storage: storage
        })

      {^ref, {:data, data}} ->
        transferred_bytes = byte_size(data) + state.transferred_bytes
        asset = Download.progress!(state.asset, %{transferred_bytes: transferred_bytes})
        {:ok, storage} = Storage.append(state.storage, data)

        handle_response(ref, req, %{
          state
          | asset: asset,
            transferred_bytes: transferred_bytes,
            storage: storage
        })

      {^ref, :done} ->
        with {:ok, _} <- Storage.commit(state.storage),
             {:ok, _} <- notify_complete(state.asset) do
          :ok
        end

      {^ref, {:error, reason}} ->
        opts =
          state
          |> Map.take([:asset, :method, :uri])
          |> Map.put(:error, reason)
          |> Enum.to_list()

        Storage.destroy(state.storage)

        {:error, HttpError.exception(opts)}
    end
  end

  defp notify_complete(asset) when asset.asset_type == :feed,
    do: Podcast.feed_download_complete(asset.id)

  defp notify_complete(_asset), do: {:error, "THIS SHOULD NOT HAVE HAPPENED"}

  defp default_headers(asset, uri) do
    headers =
      asset.headers
      |> Map.put_new("host", uri.host)
      |> Map.put_new(
        "user-agent",
        "Podbox/#{@vsn} (#{inspect(__MODULE__)}; +https://harton.dev/james/podbox)"
      )
      |> Enum.to_list()

    {:ok, headers}
  end

  defp fetch_location_header(resp_headers, state) do
    resp_headers
    |> Enum.reduce_while(:error, fn
      {"location", value}, :error -> {:halt, {:ok, value}}
      _, :error -> {:cont, :error}
    end)
    |> case do
      {:ok, location} ->
        {:ok, location}

      :error ->
        opts =
          state
          |> Map.take([:asset, :method, :req_headers, :status, :uri])
          |> Map.put(:resp_headers, resp_headers)
          |> Enum.to_list()

        {:error, BadRedirectError.exception(opts)}
    end
  end

  defp get_header(headers, header_name, default \\ nil) do
    Enum.reduce_while(headers, default, fn
      {^header_name, value}, _ -> {:halt, value}
      _, default -> {:cont, default}
    end)
  end

  defp validate_uri(uri) do
    with {:ok, uri} <- validate_uri_format(uri) do
      validate_uri_scheme(uri)
    end
  end

  defp validate_uri_format(uri) do
    case URI.new(uri) do
      {:ok, uri} -> {:ok, uri}
      {:error, invalid} -> {:error, InvalidUriError.exception(uri: uri, invalid: invalid)}
    end
  end

  defp validate_uri_scheme(uri) when uri.scheme in ["http", "https"], do: {:ok, uri}
  defp validate_uri_scheme(uri), do: {:error, UnsupportedSchemeError.exception(uri: uri)}
end