repo.ex

defmodule Recruitee.Repo do
  use Ecto.Repo, otp_app: :recruitee
  import Ecto.Query

  @doc """
  Stream query results

  Example:
      iex> Candidate
      ...> |> where([c], c.foo > 4)
      ...> |> Repo.stream(chunk_size: 300)
  """
  def stream(query, opts \\ []) do
    chunk_size = Keyword.get(opts, :chunk_size, 1000)
    on_chunk = Keyword.get(opts, :on_chunk)

    Stream.resource(
      # start with id=0
      fn -> {query, 0} end,

      # get at most `chunk_size` items
      fn {query, last_id} ->
        list = query
          |> where([e], e.id > ^last_id)
          |> limit(^chunk_size)
          |> order_by([e], e.id)
          |> all
          
        if on_chunk, do: on_chunk.(list)
  
        case List.last(list) do
          %{id: id} -> {list, {query, id}}
          nil       -> {:halt, {query, last_id}}
        end
      end,

      fn _ -> [] end
    )
  end
end