Elixir simple cursornator

simple_cursornator.ex

defmodule Cursornator do
  import Ecto.Query

  @doc """
  Stream the query with cursor

  Options

  - `:loop_query_func` - function to go to the next page
  - `:limit` - limit of entries to fetch
  - `:order_expr` - order expression for ordering the query
  
  Example
  
  |> Cursornator.stream([:field1, :field2, :id])
  """
  def stream(query, fields, opts \\ []) do
    loop_query_func = Keyword.get(opts, :loop_query_func)
    limit = Keyword.get(opts, :limit, 100)
    order_expr = Keyword.get(opts, :order_expr, Enum.map(fields, &{:asc, &1}))

    query =
      query
      |> order_by(^order_expr)
      |> limit(^limit)

    Stream.resource(
      fn -> %{values: [], has_next?: true} end,
      fn stream_opts ->
        case stream_opts do
          %{has_next?: false} ->
            {:halt, nil}

          _ ->
            query_entries(query, stream_opts[:values], fields, loop_query_func, limit)
        end
      end,
      fn _ -> nil end
    )
  end

  # Query entries with cursor
  defp query_entries(query, values, fields, loop_query_func, limit) do
    query =
      if is_nil(loop_query_func) do
        loop_query(query, fields, values)
      else
        loop_query_func.(query, fields, values)
      end

    entries = NetSuiteMiddlewareCore.Repo.all(query)

    if length(entries) == limit do
      last_entry = List.last(entries)
      new_values = Enum.map(fields, &{&1, Map.get(last_entry, &1)})

      {entries, %{values: new_values, has_next?: true}}
    else
      {entries, %{values: [], has_next?: false}}
    end
  end

  # Build condition for looping query
  defp loop_query(query, _fields, []), do: query

  defp loop_query(query, fields, values) do
    reversed_fields = Enum.reverse(fields)
    [last_field | reversed_fields] = reversed_fields
    condition = dynamic([entry], field(entry, ^last_field) > ^values[last_field])

    condition =
      Enum.reduce(reversed_fields, condition, fn field, acc ->
        value = values[field]

        dynamic(
          [entry],
          field(entry, ^field) > ^value or (field(entry, ^field) == ^value and ^acc)
        )
      end)

    where(query, ^condition)
  end
end