normalized_query.ex

defmodule Mongo.Ecto.NormalizedQuery do
  @moduledoc false

  defmodule ReadQuery do
    @moduledoc false

    defstruct coll: nil,
              pk: nil,
              params: {},
              query: %{},
              projection: %{},
              order: %{},
              fields: [],
              database: nil,
              opts: []
  end

  defmodule WriteQuery do
    @moduledoc false

    defstruct op: nil,
              coll: nil,
              query: %{},
              command: %{},
              database: nil,
              returning: [],
              pk: nil,
              opts: []
  end

  defmodule CommandQuery do
    @moduledoc false

    defstruct command: nil, database: nil, opts: []
  end

  defmodule CountQuery do
    @moduledoc false

    defstruct coll: nil, pk: nil, fields: [], query: %{}, database: nil, opts: []
  end

  defmodule AggregateQuery do
    @moduledoc false

    defstruct coll: nil, pk: nil, fields: [], pipeline: [], database: nil, opts: []
  end

  require Ecto.Query

  alias Ecto.Query
  alias Mongo.Ecto.Conversions

  defmacrop is_op(op) do
    quote do
      is_atom(unquote(op)) and unquote(op) != :^
    end
  end

  def all(original, params) do
    check_query!(original, [:limit, :offset])

    from = from(original)
    params = List.to_tuple(params)
    query = query(original, params, from)

    case projection(original, params, from) do
      {:count, fields} ->
        count(original, query, fields, params, from)

      {:find, projection, fields} ->
        find_all(original, query, projection, fields, params, from)

      {:aggregate, pipeline, fields} ->
        aggregate(original, query, pipeline, fields, params, from)
    end
  end

  defp find_all(original, query, projection, fields, params, {coll, _, pk} = from) do
    %ReadQuery{
      coll: coll,
      pk: pk,
      params: params,
      query: query,
      fields: fields,
      projection: projection,
      order: order(original, from),
      database: original.prefix,
      opts: limit_skip(original, params, from)
    }
  end

  defp count(original, query, fields, params, {coll, _, pk} = from) do
    %CountQuery{
      coll: coll,
      query: query,
      opts: limit_skip(original, params, from),
      pk: pk,
      fields: fields,
      database: original.prefix
    }
  end

  defp aggregate(original, query, pipeline, fields, params, {coll, _, pk} = from) do
    pipeline =
      limit_skip(original, params, from)
      |> Enum.map(fn
        {:limit, value} -> ["$limit": value]
        {:skip, value} -> ["$skip": value]
      end)
      |> Kernel.++(pipeline)

    pipeline = if query != %{}, do: [["$match": query] | pipeline], else: pipeline

    %AggregateQuery{
      coll: coll,
      pipeline: pipeline,
      pk: pk,
      fields: fields,
      database: original.prefix
    }
  end

  def update_all(%Query{} = original, params) do
    check_query!(original)

    params = List.to_tuple(params)
    from = from(original)
    coll = coll(from)
    query = query(original, params, from)
    command = command(:update, original, params, from)

    %WriteQuery{coll: coll, query: query, command: command, database: original.prefix}
  end

  def update_one(%{source: coll, prefix: prefix, schema: schema}, fields, filter) do
    command = command(:update, fields, primary_key(schema))
    query = query(filter, primary_key(schema))

    %WriteQuery{coll: coll, query: query, database: prefix, command: command}
  end

  def update(%{source: coll, prefix: prefix, schema: schema}, fields, filter) do
    command = command(:update, fields, primary_key(schema))
    query = query(filter, primary_key(schema))

    %WriteQuery{coll: coll, query: query, database: prefix, command: command}
  end

  def delete_all(%Query{} = original, params) do
    check_query!(original)

    params = List.to_tuple(params)
    from = from(original)
    coll = coll(from)
    query = query(original, params, from)

    %WriteQuery{coll: coll, query: query, database: original.prefix}
  end

  def delete(%{source: coll, schema: schema, prefix: prefix}, filter) do
    query = query(filter, primary_key(schema))

    %WriteQuery{coll: coll, query: query, database: prefix}
  end

  # `:raise` is the default behaviour when on_conflict is not specified. In this
  # case we assume that any conflict will be raised by the driver (e.g.
  # duplicate value on a uniquely indexed field)
  def insert(schema_meta, fields, {:raise, [], []}, returning, _opts),
    do: plain_insert(schema_meta, fields, returning)

  # When a user specifies `on_conflict: :nothing` with no conflict targets then
  # we can treat this as a plain insert.  It's expected that exceptions from the
  # driver will be suppressed elsewhere as appropriate.
  def insert(schema_meta, fields, {:nothing, [], []}, returning, _opts),
    do: plain_insert(schema_meta, fields, returning)

  def insert(
        %{source: coll, schema: schema, prefix: prefix},
        [[_ | _] | _] = docs,
        {:nothing, [], conflict_targets},
        returning,
        _opts
      ) do
    # check_conflict_targets!(schema, conflict_targets)

    pk = primary_key(schema)

    q = %WriteQuery{
      op: :update,
      coll: coll,
      database: prefix,
      command: [],
      returning: returning,
      pk: pk
    }

    docs
    |> Enum.reduce(q, fn fields, acc ->
      %{command: command} = acc

      query = query(fields, conflict_targets, pk)

      case query do
        query when query == %{} ->
          doc = fields |> value(pk, "insert")

          %{acc | op: :insert_all, command: [doc | command]}

        query ->
          update = command(:update, [], fields, pk)

          update_command = [
            query: query,
            update: update,
            upsert: true
          ]

          %{acc | command: [update_command | command]}
      end
    end)
  end

  # User specified `on_conflict: :nothing` with specified fields that must be
  # checked for conflicts.
  def insert(
        %{source: coll, schema: schema, prefix: prefix},
        fields,
        {:nothing, [], conflict_targets},
        returning,
        _opts
      ) do
    pk = primary_key(schema)

    # Use `conflict_targets` to create a query that will match any existing documents
    query = query(fields, conflict_targets, pk)

    case query do
      query when query == %{} ->
        %WriteQuery{
          op: :insert,
          coll: coll,
          command: fields |> value(pk, "insert"),
          database: prefix,
          returning: returning,
          pk: pk
        }

      query ->
        command = command(:update, [], fields, pk)

        # * `upsert: true` creates an existing document if none already exists.
        # * `return_document: :after` causes the upserted document to be returned,
        #   which we need since we need to know the ID of the inserted or updated
        #   document.  Ecto expects updates **not** to return an ID but Mongo always
        #   does (seems like a rare moment where Mongo has slightly clearer support
        #   than SQL).  It's expected that this discrepancy is massaged out
        #   elsewhere.
        opts = [
          upsert: true
        ]

        %WriteQuery{
          op: :find_one_and_update,
          coll: coll,
          query: query,
          command: command,
          database: prefix,
          returning: returning,
          pk: pk,
          opts: opts
        }
    end
  end

  # User specified a list of fields to replace and conflict targets to check.
  # This gets called if the user has specified an `on_conflict` value of
  # `:replace_all`, `{:replace, fields}` or `{:replace_all_except, fields}`
  #
  # Ecto wants us to be able to support the replacement of the `id` field which
  # MongoDB forbids, so this gets a bit tricky.  When the user wishes to replace
  # the `id` field then the only way I have found to make this work is to
  # perform a delete followed by an insert; not ideal since data loss could
  # occur.  We can't do an insert followed by a delete instead because the
  # inserted document could contain conflicts with the document that is about to
  # be deleted.
  def insert(
        %{schema: schema, source: coll, prefix: prefix},
        [[_ | _] | _] = docs,
        {[_ | _] = replace_fields, _, conflict_targets},
        returning,
        opts
      ) do
    check_supported!(conflict_targets, Keyword.get(opts, :on_conflict))

    pk = primary_key(schema)

    q = %WriteQuery{
      op: :update,
      coll: coll,
      database: prefix,
      command: [],
      returning: returning,
      pk: pk
    }

    docs
    |> Enum.reduce(q, fn fields, acc ->
      query = query(fields, conflict_targets, pk)

      %{command: command} = acc

      case query do
        query when query == %{} ->
          doc = fields |> value(pk, "insert all doc")

          %{acc | op: :insert_all, command: [doc | command]}

        query ->
          if pk in replace_fields do
            doc = fields |> value(pk, "insert_all doc")
            # another case where we have to just do an insert
            %{acc | op: :insert_all, command: [doc | command]}
          else
            {set_fields, set_on_insert_fields} =
              upsert_fields(fields, replace_fields, conflict_targets)

            update = command(:update, set_fields, set_on_insert_fields, pk)

            update_command = [
              query: query,
              update: update,
              upsert: true
            ]

            %{acc | command: [update_command | command]}
          end
      end
    end)
  end

  def insert(
        %{schema: schema} = schema_meta,
        fields,
        {[_ | _] = replace_fields, _, conflict_targets},
        returning,
        opts
      ) do
    pk = primary_key(schema)
    query = query(fields, conflict_targets, pk)

    if pk in replace_fields do
      plain_insert(schema_meta, fields, returning)
    else
      {set_fields, set_on_insert_fields} = upsert_fields(fields, replace_fields, conflict_targets)

      upsert(schema_meta, query, set_fields, set_on_insert_fields, returning, opts)
    end
  end

  def insert(
        %{source: coll, prefix: prefix},
        [[_ | _] | _] = docs,
        {%Ecto.Query{} = query, values, conflict_targets},
        returning,
        _opts
      ) do
    from = from(query)
    {_coll, _schema, pk} = from

    q = %WriteQuery{
      op: :update,
      coll: coll,
      database: prefix,
      command: [],
      returning: returning,
      pk: pk
    }

    docs
    |> Enum.reduce(q, fn fields, %{command: command} = acc ->
      # Create a query for finding existing documents based on the conflict targets
      find_query_values =
        fields |> Keyword.take(conflict_targets) |> Keyword.values() |> List.to_tuple()

      find_query =
        query
        |> filtering_conflict_targets(fields, conflict_targets)
        |> query(find_query_values, from)

      case find_query do
        find_query when find_query == %{} ->
          # Empty query, have to convert this back into an insert
          doc = fields |> value(pk, "insert all")

          %{acc | op: :insert_all, command: [doc | command]}

        find_query ->
          # If the query has specified a where then this cannot be an upsert; it's
          # expected that a non-matching query return no results (and be reported as
          # stale by Ecto)
          upsert = query.wheres == []

          update =
            command(:update, query, List.to_tuple(Keyword.values(fields) ++ values), fields, from)

          update_command = [
            query: find_query,
            update: update,
            upsert: upsert
          ]

          %{acc | command: [update_command | command]}
      end
    end)
  end

  # User specified a query to perform in the event of a conflict.
  #
  # This is achieved using the mongo `$project` operator.  Undoubtedly this
  # could be refactored to be clearer code wise but for now since `$project` is
  # only used here a long function will suffice.
  def insert(
        %{source: coll, prefix: prefix},
        fields,
        {%Ecto.Query{} = query, values, conflict_targets},
        returning,
        _opts
      ) do
    # check_conflict_targets!(schema, conflict_targets)

    from = from(query)
    {_coll, _schema, pk} = from

    # Create a query for finding existing documents based on the conflict targets
    find_query_values =
      fields |> Keyword.take(conflict_targets) |> Keyword.values() |> List.to_tuple()

    find_query =
      query
      |> filtering_conflict_targets(fields, conflict_targets)
      |> query(find_query_values, from)

    case find_query do
      find_query when find_query == %{} ->
        %WriteQuery{
          op: :insert,
          coll: coll,
          database: prefix,
          command: fields,
          returning: returning,
          pk: pk
        }

      find_query ->
        # If the query has specified a where then this cannot be an upsert; it's
        # expected that a non-matching query return no results (and be reported as
        # stale by Ecto)
        upsert = query.wheres == []

        update =
          command(:update, query, List.to_tuple(Keyword.values(fields) ++ values), fields, from)

        opts = [
          upsert: upsert,
          return_document: :after
        ]

        %WriteQuery{
          op: :find_one_and_update,
          coll: coll,
          database: prefix,
          query: find_query,
          command: update,
          returning: returning,
          pk: pk,
          opts: opts
        }
    end
  end

  defp check_supported!([], {:replace_all_except, _fields}) do
    raise """
    Upserts with `:replace_all_except` and no conflict targets are not supported since
    MongoDB does not allow post-conflict actions in a single atomic operation.

    To work around this issue either:

    * Specify a conflict target
    * Try a different on_conflict strategy
    """
  end

  defp check_supported!(_fields, _on_conflict), do: :ok

  defp plain_insert(%{source: coll, schema: schema, prefix: prefix}, fields, returning) do
    pk = primary_key(schema)

    command = command(:insert, fields, pk)

    op =
      case fields do
        [[_ | _] | _] -> :insert_all
        _ -> :insert
      end

    %WriteQuery{
      op: op,
      coll: coll,
      command: command,
      database: prefix,
      returning: returning,
      pk: pk
    }
  end

  # When both `conflict_targets and `set_on_insert` are empty we can consider this a plain insert.
  defp upsert(schema_meta, query, set_fields, [], returning, _opts)
       when query == %{} or query == [],
       do: plain_insert(schema_meta, set_fields, returning)

  defp upsert(
         %{source: coll, schema: schema, prefix: prefix},
         query,
         set_fields,
         set_on_insert_fields,
         returning,
         _opts
       ) do
    pk = primary_key(schema)

    update = command(:update, set_fields, set_on_insert_fields, pk)

    opts = [
      return_document: :after,
      upsert: true
    ]

    %WriteQuery{
      op: :find_one_and_update,
      coll: coll,
      database: prefix,
      query: query,
      command: update,
      returning: returning,
      pk: pk,
      opts: opts
    }
  end

  def command(command, opts) do
    %CommandQuery{command: command, database: Keyword.get(opts, :database)}
  end

  # Splits a list of fields into individual set and set_on_insert lists
  defp upsert_fields(fields, replace_fields, conflict_targets) do
    fields
    |> Keyword.split(replace_fields)
    |> (fn {set_fields, set_on_insert_fields} ->
          {
            set_fields |> Keyword.drop(conflict_targets),
            set_on_insert_fields ++ Keyword.take(set_fields, conflict_targets)
          }
        end).()
  end

  defp filtering_conflict_targets(%Ecto.Query{} = query, fields, conflict_targets) do
    conflict_field_filters = fields |> Keyword.take(conflict_targets)

    Ecto.Query.from(u in query, where: ^conflict_field_filters)
  end

  defp from(%Query{from: %{source: {coll, model}}}) do
    {coll, model, primary_key(model)}
  end

  defp from(%Query{from: %{source: %Ecto.SubQuery{}}}) do
    raise ArgumentError, "MongoDB does not support subqueries"
  end

  @aggregate_ops [:min, :max, :sum, :avg]
  @special_ops [:count | @aggregate_ops]

  defp projection(%Query{select: nil}, _params, _from), do: {:find, %{}, []}

  defp projection(
         %Query{select: %Query.SelectExpr{fields: fields} = _select} = query,
         params,
         from
       ) do
    projection(fields, params, from, query, %{}, [])
  end

  defp projection([], _params, _from, _query, pacc, facc), do: {:find, pacc, Enum.reverse(facc)}

  # TODO this projection function is the same as the one below with a different pattern
  defp projection(
         [{:&, _, [0]} = field | rest],
         params,
         {_, nil, _} = from,
         query,
         _pacc,
         facc
       ) do
    facc =
      case projection(rest, params, from, query, %{}, [field | facc]) do
        {:find, _, facc} ->
          facc

        _other ->
          error(
            query,
            "select clause supports only one of the special functions: `count`, `min`, `max`"
          )
      end

    {:find, %{}, facc}
  end

  defp projection(
         [{:&, _, [0, nil, _]} = field | rest],
         params,
         {_, nil, _} = from,
         query,
         _pacc,
         facc
       ) do
    # Model is nil, we want empty projection, but still extract fields
    facc =
      case projection(rest, params, from, query, %{}, [field | facc]) do
        {:find, _, facc} ->
          facc

        _other ->
          error(
            query,
            "select clause supports only one of the special functions: `count`, `min`, `max`"
          )
      end

    {:find, %{}, facc}
  end

  defp projection(
         [{:&, _, [0, nil, _]} = field | rest],
         params,
         {_, model, pk} = from,
         query,
         pacc,
         facc
       ) do
    pacc = Enum.into(model.__schema__(:fields), pacc, &{field(&1, pk), true})
    facc = [field | facc]

    projection(rest, params, from, query, pacc, facc)
  end

  defp projection(
         [{:&, _, [0, fields, _]} = field | rest],
         params,
         {_, _model, pk} = from,
         query,
         pacc,
         facc
       ) do
    pacc = Enum.into(fields, pacc, &{field(&1, pk), true})
    facc = [field | facc]

    projection(rest, params, from, query, pacc, facc)
  end

  defp projection([%Ecto.Query.Tagged{value: value} | rest], params, from, query, pacc, facc) do
    {_, model, pk} = from

    pacc = Enum.into(model.__schema__(:fields), pacc, &{field(&1, pk), true})
    facc = [{:field, pk, value} | facc]

    projection(rest, params, from, query, pacc, facc)
  end

  defp projection([{{:., _, [_, name]}, _, _} = field | rest], params, from, query, pacc, facc) do
    {_, _, pk} = from

    # Projections use names as in database, fields as in models
    pacc = Map.put(pacc, field(name, pk), true)
    facc = [{:field, name, field} | facc]
    projection(rest, params, from, query, pacc, facc)
  end

  # Keyword and interpolated fragments
  defp projection([{:fragment, _, [args]} = field | rest], params, from, query, pacc, facc)
       when is_list(args) or tuple_size(args) == 3 do
    {_, _, pk} = from

    pacc =
      args
      |> value(params, pk, query, "select clause")
      |> Enum.into(pacc)

    facc = [field | facc]

    projection(rest, params, from, query, pacc, facc)
  end

  defp projection([{:count, _, [_]} = field], _params, _from, _query, pacc, _facc)
       when pacc == %{} do
    {:count, [{:field, :value, field}]}
  end

  defp projection([{:count, _, [name, :distinct]} = field], _params, from, query, _pacc, _facc) do
    {_, _, pk} = from
    name = field(name, pk, query, "select clause")
    field = {:field, :value, field}

    {:aggregate, [["$group": [_id: "$#{name}"]], ["$group": [_id: nil, value: ["$sum": 1]]]],
     [field]}
  end

  defp projection([{op, _, [name]} = field], _params, from, query, pacc, _facc)
       when pacc == %{} and op in @aggregate_ops do
    {_, _, pk} = from
    name = field(name, pk, query, "select clause")
    field = {:field, :value, field}
    {:aggregate, [["$group": [_id: nil, value: [{"$#{op}", "$#{name}"}]]]], [field]}
  end

  defp projection([{op, _, _} | _rest], _params, _from, query, _pacc, _facc)
       when op in @special_ops do
    error(
      query,
      "select clause supports only one of the special functions: `count`, `min`, `max`"
    )
  end

  defp projection([{op, _, _} | _rest], _params, _from, query, _pacc, _facc) when is_op(op) do
    error(query, "select clause")
  end

  defp limit_skip(%Query{limit: limit, offset: offset} = query, params, {_, _, pk}) do
    [
      limit: offset_limit(limit, params, pk, query, "limit clause"),
      skip: offset_limit(offset, params, pk, query, "offset clause")
    ]
    |> Enum.reject(&is_nil(elem(&1, 1)))
  end

  defp coll({coll, _model, _pk}), do: coll

  defp query(%Query{wheres: wheres} = query, params, {_coll, _model, pk}) do
    wheres
    |> Enum.map(fn %Query.BooleanExpr{expr: expr} ->
      pair(expr, params, pk, query, "where clause")
    end)
    |> :lists.flatten()
    |> merge_keys(query, "where clause")
    |> map_unless_empty
  end

  defp query([{_, _} | _] = fields, keys, pk) do
    fields
    |> Keyword.take(keys)
    |> query(pk)
  end

  defp query(filter, pk) do
    filter |> value(pk, "where clause") |> map_unless_empty
  end

  defp order(%Query{order_bys: order_bys} = query, {_coll, _model, pk}) do
    order_bys
    |> Enum.flat_map(fn %Query.QueryExpr{expr: expr} ->
      Enum.map(expr, &order_by_expr(&1, pk, query))
    end)
    |> map_unless_empty
  end

  defp command(:update, %Query{updates: updates} = query, params, {_coll, _model, pk}) do
    updates
    |> Enum.flat_map(fn %Query.QueryExpr{expr: expr} ->
      Enum.map(expr, fn {key, value} ->
        value = value |> value(params, pk, query, "update clause")
        {update_op(key, query), value}
      end)
    end)
    |> merge_keys(query, "update clause")
  end

  defp command(:update, [], set_on_insert_fields, pk), do: set_on_insert(set_on_insert_fields, pk)

  defp command(:update, set_fields, set_on_insert_fields, pk) do
    # Set fields are the same as doing a plain update with `$set`
    set_command = command(:update, set_fields, pk)

    # These fields are only applied if the operation results in an insert.
    set_on_insert_command = set_on_insert(set_on_insert_fields, pk)

    set_command ++ set_on_insert_command
  end

  defp command(
         :update,
         %Query{} = query,
         params,
         set_on_insert_fields,
         {_coll, _model, pk} = from
       ) do
    update = command(:update, query, params, from)

    set_fields = Map.get(update, :"$set", [])
    inc_fields = Map.get(update, :"$inc", [])

    # The behaviour we want is for different values to be set on the database
    # depending on whether this is an insert or an update.  In Mongo land we
    # might tend to reach for `$set` and `$setOnInsert`, however Mongo disallows
    # us from having the same fields in both `$set` and `$setOnInsert`, so that
    # won't work.
    #
    # The workaround implemented below is to use the `$project` aggregate
    # operator along with `$cond`.  `$cond` allows us to check for the presence
    # of an `_id` field, so we can `$project` different fields depending on
    # whether this was an insert or an update.

    # ID is a special case
    pkk = :_id
    pkv = "$_id"

    projection = %{}

    # If an existing document matches `find_query`, then `_id` will be present.
    # If we specify an `_id` in the projection of an **existing** document then
    # Mongo complains, so we must omit it in this case.  There's a special
    # variable `$REMOVE` for this purpose which tells Mongo to omit that field
    # from the projection.
    projection =
      projection
      |> Map.put(pkk, %{"$cond": ["$#{pkk}", "$REMOVE", pkv]})

    # set_fields are either set to the value from the query (in the event of an
    # update) or the value in fields (in the event of an insert)
    projection =
      set_fields
      |> Enum.reduce(projection, fn {k, v}, acc ->
        Map.put(acc, k, %{"$cond": ["$#{pkk}", v, set_on_insert_fields[k]]})
      end)

    # inc_fields are incremented in the case of an update or set from fields in
    # the event of an insert
    projection =
      inc_fields
      |> Enum.reduce(projection, fn {field, increment_by}, acc ->
        Map.put(acc, field, %{
          "$cond": ["$#{pkk}", %{"$sum": ["$#{field}", increment_by]}, "$#{field}"]
        })
      end)

    # All other fields are just plain "set" operation.  Because we don't want
    # them to conflict with the set vs. set_on_insert operations above, we can
    # use `Map.put_new` to ensure there are no duplicates.
    projection =
      set_on_insert_fields
      |> value(pk, "projection")
      |> Enum.reduce(projection, fn {k, v}, acc ->
        Map.put_new(acc, k, v)
      end)

    [
      %{
        "$project": projection
      }
    ]
  end

  defp command(:insert, document, pk) do
    document
    |> value(pk, "insert command")
    |> map_unless_empty
  end

  defp command(:update, values, pk) do
    ["$set": values |> value(pk, "update command") |> map_unless_empty]
  end

  defp set_on_insert(fields, pk) do
    [
      "$setOnInsert": fields |> value(pk, "update command (set on insert)") |> map_unless_empty()
    ]
  end

  defp offset_limit(nil, _params, _pk, _query, _where), do: nil

  defp offset_limit(%Query.QueryExpr{expr: expr}, params, pk, query, where),
    do: value(expr, params, pk, query, where)

  defp offset_limit(%Query.LimitExpr{expr: expr}, params, pk, query, where),
    do: value(expr, params, pk, query, where)

  defp primary_key(nil), do: nil

  defp primary_key(schema) do
    case schema.__schema__(:primary_key) do
      [] ->
        nil

      [pk] ->
        pk

      keys ->
        raise ArgumentError,
              "MongoDB adapter does not support multiple primary keys " <>
                "and #{inspect(keys)} were defined in #{inspect(schema)}."
    end
  end

  defp order_by_expr({:asc, expr}, pk, query), do: {field(expr, pk, query, "order clause"), 1}
  defp order_by_expr({:desc, expr}, pk, query), do: {field(expr, pk, query, "order clause"), -1}

  @maybe_disallowed ~w(distinct lock joins group_bys havings limit offset)a
  @query_empty_values %Ecto.Query{} |> Map.take(@maybe_disallowed)

  defp check_query!(query, allow \\ []) do
    @query_empty_values
    |> Map.drop(allow)
    |> Enum.each(fn {element, empty} ->
      check(
        Map.get(query, element),
        empty,
        query,
        "MongoDB adapter does not support #{element} clause in this query"
      )
    end)
  end

  defp check(expr, expr, _, _), do: nil
  defp check(_, _, query, message), do: raise(Ecto.QueryError, query: query, message: message)

  defp value(expr, pk, place) do
    case Conversions.from_ecto_pk(expr, pk) do
      {:ok, value} -> value
      :error -> error(place)
    end
  end

  defp value(expr, params, pk, query, place) do
    case Conversions.inject_params(expr, params, pk) do
      {:ok, value} -> value
      :error -> error(query, place)
    end
  end

  defp field(pk, pk), do: :_id
  defp field(key, _), do: key

  defp field({{:., _, [{:&, _, [0]}, field]}, _, []}, pk, _query, _place), do: field(field, pk)
  defp field(_expr, _pk, query, place), do: error(query, place)

  defp map_unless_empty([]), do: %{}
  defp map_unless_empty(list), do: list

  defp merge_keys(keyword, query, place) do
    Enum.reduce(keyword, %{}, fn {key, value}, acc ->
      Map.update(acc, key, value, fn
        old when is_list(old) -> old ++ value
        _ -> error(query, place)
      end)
    end)
  end

  update_ops = [set: :"$set", inc: :"$inc", push: :"$push", pull: :"$pull"]

  Enum.map(update_ops, fn {key, op} ->
    def update_op(unquote(key), _query), do: unquote(op)
  end)

  def update_op(_, query), do: error(query, "update clause")

  binary_ops = [>: :"$gt", >=: :"$gte", <: :"$lt", <=: :"$lte", !=: :"$ne", in: :"$in"]
  bool_ops = [and: :"$and", or: :"$or"]

  @binary_ops Keyword.keys(binary_ops)
  @bool_ops Keyword.keys(bool_ops)

  Enum.map(binary_ops, fn {op, mongo_op} ->
    defp binary_op(unquote(op)), do: unquote(mongo_op)
  end)

  Enum.map(bool_ops, fn {op, mongo_op} ->
    defp bool_op(unquote(op)), do: unquote(mongo_op)
  end)

  defp mapped_pair_or_value({op, _, _} = tuple, params, pk, query, place) when is_op(op) do
    List.wrap(pair(tuple, params, pk, query, place))
  end

  defp mapped_pair_or_value(value, params, pk, query, place) do
    value(value, params, pk, query, place)
  end

  defp pair({op, _, args}, params, pk, query, place) when op in @bool_ops do
    args = Enum.map(args, &mapped_pair_or_value(&1, params, pk, query, place))
    {bool_op(op), args}
  end

  defp pair({:is_nil, _, [expr]}, _, pk, query, place) do
    {field(expr, pk, query, place), nil}
  end

  defp pair({:==, _, [left, right]}, params, pk, query, place) do
    {field(left, pk, query, place), value(right, params, pk, query, place)}
  end

  defp pair({:in, _, [left, {:^, _, [0, 0]}]}, _params, pk, query, place) do
    {field(left, pk, query, place), ["$in": []]}
  end

  defp pair({:in, _, [left, {:^, _, [ix, len]}]}, params, pk, query, place) do
    args =
      ix..(ix + len - 1)
      |> Enum.map(&elem(params, &1))
      |> Enum.map(&value(&1, params, pk, query, place))

    {field(left, pk, query, place), ["$in": args]}
  end

  defp pair({:in, _, [lhs, {{:., _, _}, _, _} = rhs]}, params, pk, query, place) do
    {field(rhs, pk, query, place), value(lhs, params, pk, query, place)}
  end

  defp pair({op, _, [left, right]}, params, pk, query, place) when op in @binary_ops do
    {field(left, pk, query, place), [{binary_op(op), value(right, params, pk, query, place)}]}
  end

  defp pair({:not, _, [{:in, _, [left, {:^, _, [ix, len]}]}]}, params, pk, query, place) do
    args =
      ix..(ix + len - 1)
      |> Enum.map(&elem(params, &1))
      |> Enum.map(&value(&1, params, pk, query, place))

    {field(left, pk, query, place), ["$nin": args]}
  end

  defp pair({:not, _, [{:in, _, [left, right]}]}, params, pk, query, place) do
    {field(left, pk, query, place), ["$nin": value(right, params, pk, query, place)]}
  end

  defp pair({:not, _, [{:is_nil, _, [expr]}]}, _, pk, query, place) do
    {field(expr, pk, query, place), ["$ne": nil]}
  end

  defp pair({:not, _, [{:==, _, [left, right]}]}, params, pk, query, place) do
    {field(left, pk, query, place), ["$ne": value(right, params, pk, query, place)]}
  end

  defp pair({:not, _, [expr]}, params, pk, query, place) do
    {:"$not", [pair(expr, params, pk, query, place)]}
  end

  defp pair({:^, _, _} = expr, params, pk, query, place) do
    case value(expr, params, pk, query, place) do
      bool when is_boolean(bool) ->
        boolean_query_hack_pair(bool)

      _value ->
        error(query, place)
    end
  end

  # Keyword or embedded fragment
  defp pair({:fragment, _, [args]}, params, pk, query, place)
       when is_list(args) or tuple_size(args) == 3 do
    value(args, params, pk, query, place)
  end

  defp pair(bool, _params, _pk, _query, _place) when is_boolean(bool) do
    boolean_query_hack_pair(bool)
  end

  defp pair(_expr, _params, _pk, query, place) do
    error(query, place)
  end

  defp boolean_query_hack_pair(bool) do
    {:_id, ["$exists": bool]}
  end

  defp error(query, place) do
    raise Ecto.QueryError,
      query: query,
      message: "Invalid expression for MongoDB adapter in #{place}"
  end

  defp error(place) do
    raise ArgumentError, "Invalid expression for MongoDB adapter in #{place}"
  end
end