   """
   def scrape_from_bigquery() do
     Logger.info "Getting repositories from BigQuery to scrape"
-    Krihelinator.Periodic.BigQuery.query()
-    |> async_scrape()
-    |> handle_scraped_data()
+    names = Krihelinator.Periodic.BigQuery.query()
+    async_handle(names)
   end
 
   @doc """

    end
  end

  @doc """
  Persist the scraped data and log results statistics.
  """
  def handle_scraped_data(data_stream) do
    data_stream
    |> Stream.map(fn
      {:ok, data} ->
        Repo.update_or_create_from_data(GH.Repo, data, by: :name)
      otherwise ->
        otherwise
    end)
    |> Stream.map(&simplify_result/1)
    |> Enum.reduce(%{}, fn x, acc -> Map.update(acc, x, 1, &(&1 + 1)) end)
    |> Enum.each(&log_aggregated_results/1)
  end

  def simplify_result({:error, %Ecto.Changeset{}}), do: :validation_error

    query = from(r in GH.Repo, where: r.dirty, select: r.name)
    names = Repo.all(query)
    Logger.info "Rescraping #{length(names)} still dirty repositories"
    async_handle(names)
  end

  @async_params [max_concurrency: 20, timeout: :infinity]

  @doc """
  Scrape and persist repositories concurrently with `Task.async_stream`.
  """
  def async_handle(names, extra_params \\ []) do
    names
    |> Task.async_stream(&handle(&1, extra_params), @async_params)
    |> Stream.map(fn {:ok, result} -> result end)
    |> Enum.reduce(%{}, fn x, acc -> Map.update(acc, x, 1, &(&1 + 1)) end)
    |> Enum.each(&log_aggregated_results/1)
  end

  @doc """
  Scrape and persist a single repository by name.
  """
  def handle(name, extra_params) do
    name
    |> scrape_with_retries()
    |> put_extra_params(extra_params)
    |> persist()
    |> simplify_result()
  end

  @to_retry ~w(github_server_error timeout)a
