ecto_adapters_dynamodb

defmodule Ecto.Adapters.DynamoDB.Query do
  @moduledoc """
  Some query wrapper functions for helping us query dynamo db. Selects indexes to use, etc.
  Not to be confused with `Ecto.Query`.
  """

  import Ecto.Adapters.DynamoDB.Info

  @typep key :: String.t
  @typep table_name :: String.t
  @typep query_op :: :== | :> | :< |:>= | :<= | :is_nil | :between | :begins_with | :in
  @typep boolean_op :: :and | :or
  @typep match_clause :: {term, query_op}
  @typep search_clause :: {key, match_clause} | {boolean_op, [search_clause]}
  @typep search :: [search_clause]
  @typep dynamo_response :: %{required(String.t) => term}
  @typep query_opts :: [{atom(), any()}]

  # parameters for get_item: 
  # TABLE_NAME::string,
  # [{LOGICAL_OP::atom, [{ATTRIBUTE::string, {VALUE::string, OPERATOR::atom}}]} | {ATTRIBUTE::string, {VALUE::string, OPERATOR::atom}}]
  #

  # Repo.all(model), provide cached results for tables designated in :cached_tables
  @spec get_item(table_name, search, keyword) :: dynamo_response | no_return
  def get_item(table, search, opts) when search == [], do: maybe_scan(table, search, opts)

  # Regular queries
  def get_item(table, search, opts) do
    parsed_index = case get_best_index!(table, search, opts) do
      # Primary key without range
      {:primary, [_idx] = idxs} ->
        {:primary, idxs}
      {:primary, idxs} ->
        {_, op2} = deep_find_key(search, Enum.at(idxs, 1))
          # Maybe query on composite primary
          if op2 in [:between, :begins_with, :<, :>, :<=, :>=] do
            {:primary_partial, idxs}
          else
            {:primary, idxs}
          end
      parsed -> parsed
    end

    results = case parsed_index do
      # primary key based lookup uses the efficient 'get_item' operation
      {:primary, indexes} = index ->
        {hash_values, op} = deep_find_key(search, hd indexes)

        if op == :in do
          # DynamoDB will reject an entire batch get query if the query is for more than 100 records, so these need to be batched.
          # https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchGetItem.html
          batch_get_item_limit = 100
          responses_element = "Responses"
          unprocessed_keys_element = "UnprocessedKeys"
          response_map = %{responses_element => %{table => []}, unprocessed_keys_element => %{}} # The default format of the response from Dynamo.

          Enum.chunk_every(hash_values, batch_get_item_limit)
          |> Enum.reduce(response_map, fn(hash_batch, acc) ->
            batched_search = make_batched_search(search, hash_batch) # Modify the 'search' arg so that it only contains values from the current hash_batch.

            %{^responses_element => %{^table => results}, ^unprocessed_keys_element => unprocessed_key_map} =
              ExAws.Dynamo.batch_get_item(construct_batch_get_item_query(table, indexes, hash_batch, batched_search, construct_opts(:get_item, opts))) |> ExAws.request!

            Kernel.put_in(acc, [responses_element, table], acc[responses_element][table] ++ results)
            |> maybe_put_unprocessed_keys(unprocessed_key_map, table, unprocessed_keys_element)
          end)
        else
          # https://hexdocs.pm/ex_aws/ExAws.Dynamo.html#get_item/3
          query = construct_search(index, search, opts)
          ExAws.Dynamo.get_item(table, query, construct_opts(:get_item, opts)) |> ExAws.request!
        end

      # secondary index based lookups need the query functionality. 
      index when is_tuple(index) ->
        index_fields = get_hash_range_key_list(index)
        {hash_values, op} = deep_find_key(search, hd index_fields)
        # https://hexdocs.pm/ex_aws/ExAws.Dynamo.html#query/2

        query = construct_search(index, search, opts)

        do_fetch_recursive = fn(qry) ->
          fetch_recursive(&ExAws.Dynamo.query/2, table, qry, parse_recursive_option(:query, opts), %{})
        end

        if op == :in do
          responses_element = "Responses"
          response_map = %{responses_element => %{table => []}}

          Enum.reduce(hash_values, response_map, fn(hash_value, acc) ->
            # When receiving a list of values to query on, construct a custom query for each of those values to pass into do_fetch_recursive/1.
            %{"Items" => items} = Kernel.put_in(query, [:expression_attribute_values, :hash_key], hash_value)
                                  |> (do_fetch_recursive).()

            Kernel.put_in(acc, [responses_element, table], acc[responses_element][table] ++ items)
          end)
        else
          do_fetch_recursive.(query)
        end
      :scan ->
        maybe_scan(table, search, opts)
    end

    filter(results, search) # index may have had more fields than the index did, thus results need to be trimmed.
  end

  # In the case of a partial query on a composite key secondary index, the value of index in get_item/2 will be a three-element tuple, ex. {:secondary_partial, "person_id_entity", ["person_id"]}.
  # Otherwise, we can expect it to be a two-element tuple.
  defp get_hash_range_key_list({_index_type, index_name, index_fields}), do: get_hash_range_key_list({index_name, index_fields})
  defp get_hash_range_key_list({_index_name, index_fields}), do: index_fields

  # If a batch_get_item request returns unprocessed keys, update the accumulator with those values.
  defp maybe_put_unprocessed_keys(acc, unprocessed_key_map, _table, _unprocessed_keys_element) when unprocessed_key_map == %{}, do: acc
  defp maybe_put_unprocessed_keys(acc, unprocessed_key_map, table, unprocessed_keys_element) do
    if Map.has_key?(acc[unprocessed_keys_element], table) do
      keys_element = "Keys"
      Kernel.put_in(acc, [unprocessed_keys_element, table, keys_element], acc[unprocessed_keys_element][table][keys_element] ++ unprocessed_key_map[table][keys_element])
    else
      Map.put(acc, unprocessed_keys_element, unprocessed_key_map)
    end
  end

  # The initial 'search' arg will have a list of all of the values being queried for;
  # when passing this data to construct_batch_get_item_query/5 during a batched operation,
  # use a modified form of the 'search' arg that contains only the values from the current batch.
  defp make_batched_search([and: [range_query, {hash_key, {_vals, op}}]], hash_batch), do: [{hash_key, {hash_batch, op}}, range_query]
  defp make_batched_search([{index, {_vals, op}}], hash_batch), do: [{index, {hash_batch, op}}]

  @doc """
  Returns an atom, :scan or :query, specifying whether the current search will be a DynamoDB scan or a query.
  """
  def scan_or_query?(table, search) do
    if get_best_index!(table, search) == :scan, do: :scan, else: :query
  end


  # we've a list of fields from an index that matches (some of) the search fields, so construct
  # a DynamoDB search criteria map with only the given fields and their search objects!
  @spec construct_search({:primary | :primary_partial | nil | String.t, [String.t]}, search, keyword) :: keyword
  @spec construct_search({:secondary_partial, String.t, [String.t]}, search, keyword) :: keyword
  def construct_search({:primary, index_fields}, search, opts), do: construct_search(%{}, index_fields, search, opts)
  def construct_search({:primary_partial, index_fields}, search, opts) do
    # do not provide index_name for primary partial
    construct_search({nil, index_fields}, search, opts)
  end
  def construct_search({index_name, index_fields}, search, opts) do
    # Construct a DynamoDB FilterExpression (since it cannot be provided blank but may be,
    # we merge it with the full query)
    {filter_expression_tuple, expression_attribute_names, expression_attribute_values} = construct_filter_expression(search, index_fields)

    updated_ops = construct_opts(:query, opts)

    # :primary_partial might not provide an index name
    criteria = if index_name != nil, do: [index_name: index_name], else: []
    criteria ++ case index_fields do
      [hash, range] ->
        {hash_val, _op} = deep_find_key(search, hash)
        {range_expression, range_attribute_names, range_attribute_values} = construct_range_params(range, deep_find_key(search, range))
        [
          # We need ExpressionAttributeNames when field-names are reserved, for example "name" or "role"
          key_condition_expression: "##{hash} = :hash_key AND #{range_expression}",
          expression_attribute_names: Enum.reduce([%{"##{hash}" => hash}, range_attribute_names, expression_attribute_names], &Map.merge/2),
          expression_attribute_values: [hash_key: hash_val] ++ range_attribute_values ++ expression_attribute_values,
        ] ++ filter_expression_tuple ++ updated_ops

      [hash] ->
        {hash_val, _op} = deep_find_key(search, hash)
        [
          key_condition_expression: "##{hash} = :hash_key",
          expression_attribute_names: Map.merge(%{"##{hash}" => hash}, expression_attribute_names),
          expression_attribute_values: [hash_key: hash_val] ++ expression_attribute_values,
        ] ++ filter_expression_tuple ++ updated_ops
      
    end
  end

  def construct_search({:secondary_partial, index_name , index_fields}, search, opts) do
    construct_search({index_name, index_fields}, search, opts)
  end

  defp construct_search(criteria, [], _, _), do: criteria
  defp construct_search(criteria, [index_field|index_fields], search, opts) do
    Map.put(criteria, index_field, elem(deep_find_key(search, index_field), 0)) |> construct_search(index_fields, search, opts)
  end


  @spec construct_range_params(key, match_clause) :: {String.t, %{required(String.t) => key}, keyword()}
  defp construct_range_params(range, {[range_start, range_end], :between}) do
    {"##{range} between :range_start and :range_end", %{"##{range}" => range}, [range_start: range_start, range_end: range_end]} 
  end
  defp construct_range_params(range, {prefix, :begins_with}) do
    {"begins_with(##{range}, :prefix)", %{"##{range}" => range}, [prefix: prefix]} 
  end
  defp construct_range_params(range, {range_val, :==}) do
    {"##{range} = :range_key", %{"##{range}" => range}, [range_key: range_val]}
  end
  defp construct_range_params(range, {range_val, op}) when op in [:<, :>, :<=, :>=] do
    {"##{range} #{to_string(op)} :range_key", %{"##{range}" => range}, [range_key: range_val]}
  end


  @spec construct_opts(atom, keyword) :: keyword
  defp construct_opts(query_type, opts) do
    take_opts = case query_type do
      :get_item -> [:consistent_read]
      :query -> [:exclusive_start_key, :limit, :scan_index_forward, :consistent_read]
    end
    case opts[:projection_expression] do
      nil -> [select: opts[:select] || :all_attributes]
      _   -> Keyword.take(opts, [:projection_expression])
    end ++ Keyword.take(opts, take_opts)
  end


  # returns a tuple: {filter_expression_tuple, expression_attribute_names, expression_attribute_values}
  @spec construct_filter_expression(search, [String.t]) :: {[filter_expression: String.t], map, keyword}
  defp construct_filter_expression(search, index_fields) do
    # We can only construct a FilterExpression on attributes not in key-conditions.
    non_indexed_filters = collect_non_indexed_search(search, index_fields, [])

    case non_indexed_filters do
      [] -> {[], %{}, []}
      _  ->
        {filter_expression_list, expression_attribute_names, expression_attribute_values} = 
          build_filter_expression_data(non_indexed_filters, {[], %{}, %{}})

        {[filter_expression: Enum.join(filter_expression_list, " and ")],
         expression_attribute_names,
         Enum.into(expression_attribute_values, [])}
    end
  end


  # Recursively strip out the fields for key-conditions; they could be mixed with non key-conditions.
  # TODO: this may be redundant - the indexed fields can just be skipped during the expression construction
  @spec collect_non_indexed_search(search, [String.t], search) :: search
  defp collect_non_indexed_search([], _index_fields, acc), do: acc
  defp collect_non_indexed_search([search_clause | search_clauses], index_fields, acc) do
    case search_clause do
      {field, {_val, _op}} = complete_tuple when not(field in [:and, :or]) ->
        if Enum.member?(index_fields, field),
        do: collect_non_indexed_search(search_clauses, index_fields, acc),
        else: collect_non_indexed_search(search_clauses, index_fields, [complete_tuple | acc])

      {logical_op, deeper_clauses} when logical_op in [:and, :or] ->
        filtered_clauses = collect_non_indexed_search(deeper_clauses, index_fields, [])
        # don't keep empty logical_op groups
        if filtered_clauses == [],
        do: collect_non_indexed_search(search_clauses, index_fields, acc),
        else: collect_non_indexed_search(search_clauses, index_fields, [{logical_op, filtered_clauses} | acc])
    end
  end

  # Recursively reconstruct parentheticals
  @type expression_data_acc :: {[String.t], map, map}
  @spec build_filter_expression_data(search, expression_data_acc) :: expression_data_acc
  defp build_filter_expression_data([], acc), do: acc
  defp build_filter_expression_data([expr | exprs], {filter_exprs, attr_names, attr_values}) do
    case expr do
      # a list of lookup fields; iterate.
      {field, {val, op} = val_op_tuple} = complete_tuple when is_tuple(val_op_tuple) ->
        updated_filter_exprs = [construct_conditional_statement(complete_tuple) | filter_exprs]
        updated_attr_names = Map.merge(%{"##{field}" => field}, attr_names)
        updated_attr_values = Map.merge(format_expression_attribute_value(field, val, op), attr_values) 

        build_filter_expression_data(exprs, {updated_filter_exprs, updated_attr_names, updated_attr_values})

      {logical_op, exprs_list} when is_list(exprs_list) ->
        {deeper_filter_exprs, deeper_attr_names, deeper_attr_values} = build_filter_expression_data(exprs_list, {[], %{}, %{}})
        # We don't parenthesize only one expression
        updated_filter_exprs = case deeper_filter_exprs do
          [one_expression] ->
            [one_expression | filter_exprs]
          many_expressions  ->
            ["(" <> Enum.join(many_expressions, " #{to_string(logical_op)} ") <> ")" | filter_exprs]
        end
        updated_attr_names = Map.merge(deeper_attr_names, attr_names)
        updated_attr_values = Map.merge(deeper_attr_values, attr_values)

        build_filter_expression_data(exprs, {updated_filter_exprs, updated_attr_names, updated_attr_values})
    end
  end

  @spec format_expression_attribute_value(key, term, query_op | [query_op]) :: map
  defp format_expression_attribute_value(field, _val, :is_nil), do: %{String.to_atom(field <> "_val") => nil}
  defp format_expression_attribute_value(field, val, :in) do
    {result, _count} = Enum.reduce(val, {%{}, 1}, fn (v, {acc, count}) ->
      {Map.merge(acc, %{String.to_atom(field <> "_val#{to_string(count)}") => v}), count + 1}
    end)
    result
  end
  # double op
  defp format_expression_attribute_value(field, [val1, val2], [_op1, _op2]) do
    %{String.to_atom(field <> "_val1") => val1, String.to_atom(field <> "_val2") => val2}
  end
   defp format_expression_attribute_value(field, [start_val, end_val], :between) do
    %{String.to_atom(field <> "_start_val") => start_val, String.to_atom(field <> "_end_val") => end_val}
  end
  defp format_expression_attribute_value(field, val, _op), do: %{String.to_atom(field <> "_val") => val}


  # double op (neither of them ought be :==)
  @spec construct_conditional_statement({key, {term, query_op} | {[term], [query_op]}}) :: String.t
  defp construct_conditional_statement({field, {[_val1, _val2], [op1, op2]}}) do
    "##{field} #{to_string(op1)} :#{field <> "_val1"} and ##{field} #{to_string(op2)} :#{field <> "_val2"}"
  end  
  defp construct_conditional_statement({field, {_val, :is_nil}}) do
    "(##{field} = :#{field <> "_val"} or attribute_not_exists(##{field}))"
  end
  defp construct_conditional_statement({field, {val, :in}}) do
    {result, _count} = Enum.reduce(val, {[], 1}, fn (_val, {acc, count}) ->
      {[":#{field}_val#{to_string(count)}" | acc], count + 1}
    end)    
    "(##{field} in (#{Enum.join(result, ",")}))"
  end
  defp construct_conditional_statement({field, {_val, :==}}) do
    "##{field} = :#{field <> "_val"}"
  end
  defp construct_conditional_statement({field, {_val, op}}) when op in [:<, :>, :<=, :>=] do
    "##{field} #{to_string(op)} :#{field <> "_val"}"
  end
  defp construct_conditional_statement({field, {[_start_val, _end_val], :between}}) do
    "##{field} between :#{field <> "_start_val"} and :#{field <> "_end_val"}"
  end
  defp construct_conditional_statement({field, {_val, :begins_with}}) do
    "begins_with(##{field}, :#{field}_val)"
  end

  defp construct_batch_get_item_query(table, indexes, hash_values, search, opts) do
    take_opts = Keyword.take(opts, [:consistent_read, :projection_expression])
    keys = case indexes do
      [hash_key] -> 
        Enum.map(hash_values, fn hash_value -> [{String.to_atom(hash_key), hash_value}] end)

      [hash_key, range_key] ->
        {range_values, :in} = deep_find_key(search, range_key)
        zipped = Enum.zip(hash_values, range_values)
        Enum.map(zipped, fn {hash_value, range_value} ->
          [{String.to_atom(hash_key), hash_value}, {String.to_atom(range_key), range_value}]
        end)
    end

    %{table => [keys: keys] ++ take_opts}
  end


  # TODO: Given the search criteria, filter out other results that were caught in the
  # index read. TODO: Can we do this on the server side dynamo query instead?
  # see: http://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_Query.html#DDB-Query-request-FilterExpression
  # note, this doesn't save us read capacity, but DOES reduce the result set and parsing over the wire.
  @spec filter(dynamo_response, search) :: dynamo_response
  def filter(results, _search), do: results



  @doc """
  Given a map with a search criteria, finds the best index to search against it.
  Returns a tuple indicating whether it's a primary key index, or a secondary index.
  To query against a secondary index in Dynamo, we NEED to have it's index name,
  so secondary indexes are returned as a tuple with the field name, whilst
  the primary key uses the atom :primary to distinguish it.

    {:primary, [indexed_fields_list]} | {"index_name", [indexed_fields_list]}

  Exception if the index doesn't exist.
  """
  @spec get_best_index(table_name, search, query_opts) :: :not_found | {:primary, [String.t]} | {:primary_partial, [String.t]} | {String.t, [String.t]} | {:secondary_partial, String.t, [String.t]} | no_return
  def get_best_index(tablename, search, opts) do
    case get_matching_primary_index(tablename, search) do
      # if we found a primary index with hash+range match, it's probably the best index.
      {:primary, _} = index -> index

      # we've found a primary hash index, but lets check if there's a more specific
      # secondary index with hash+sort available...
      {:primary_partial, _primary_hash} = index ->
        case get_matching_secondary_index(tablename, search, opts) do
          {_, [_, _]} = sec_index -> sec_index  # we've found a better, more specific index.
          _                       -> index      # :not_found, or any other hash? default back to the primary.
        end

      # no primary found, so try for a secondary.
      :not_found -> get_matching_secondary_index(tablename, search, opts)
    end
  end


  @doc """
  Same as get_best_index, but refers to a scan option on failure
  """
  def get_best_index!(tablename, search, opts \\ []) do
    case get_best_index(tablename, search, opts) do
      :not_found -> :scan
      index      -> index
    end
  end


  @doc """
  Given a search criteria of 1 or more fields, we try find out if the primary key is a
  good match and can be used to forfill this search. Returns the tuple 
    {:primary, [hash] | [hash, range]}
  or 
    :not_found
  """
  def get_matching_primary_index(tablename, search) do
    primary_key = primary_key!(tablename)

    case match_index(primary_key, search) do
      # We found a full primary index
      {:primary, _} = index -> index
      # We might be able to use a range query for all results with
      # the hash part, such as all circle_member with a specific person_id (all a user's circles).
      :not_found            -> match_index_hash_part(primary_key, search)
    end
  end


  @doc """
  Given a keyword list containing search field, value and operator to search for (which may also be nested under logical operators, e.g., `[{"id", {"franko", :==}}]`, or `[{:and, [{"circle_id", {"123", :==}}, {"person_id", {"abc", :>}}]}`), will return the dynamo db index description that will help us match this search. return :not_found if no index is found.

  Returns a tuple of {"index_name", [ hash_key or hash,range_key]]} or :not_found
  TODO: Does not help with range queries. -> The match_index_hash_part function is
    beginning to address this.
  """
  def get_matching_secondary_index(tablename, search, opts) do
    secondary_indexes = tablename |> secondary_indexes()

    # A user may provide an :index opt in a query, in which case we will prioritize choosing that index.
    case opts[:index] do
      nil            -> find_best_match(secondary_indexes, search, :not_found)
      index_option ->
        case maybe_select_index_option(index_option, secondary_indexes) do
          nil   -> index_option_error(index_option, secondary_indexes)
          index -> index
        end
    end
  end

  defp maybe_select_index_option(index_option, secondary_indexes)
  when is_atom(index_option) do
    index_option
    |> Atom.to_string()
    |> maybe_select_index_option(secondary_indexes)
  end
  defp maybe_select_index_option(index_option, secondary_indexes), do:
    Enum.find(secondary_indexes, fn({name, _keys}) -> name == index_option end)

  @spec index_option_error(String.t | atom(), [{String.t, [String.t]}]) :: no_return
  defp index_option_error(_index_option, []) do
    raise ArgumentError, message: "#{inspect __MODULE__}.get_matching_secondary_index/3 error: :index option does not match existing secondary index names."
  end
  defp index_option_error(index_option, secondary_indexes) do
    index_option = if is_atom(index_option), do: Atom.to_string(index_option), else: index_option
    {nearest_index_name, jaro_distance} = secondary_indexes
                                          |> Enum.map(fn {name, _keys} -> {name, String.jaro_distance(index_option, name)} end)
                                          |> Enum.max_by(fn {_name, jaro_distance} -> jaro_distance end)

    case jaro_distance >= 0.75 do
      true  -> raise ArgumentError, message: "#{inspect __MODULE__}.get_matching_secondary_index/3 error: :index option does not match existing secondary index names. Did you mean #{nearest_index_name}?"
      false -> index_option_error(index_option, [])
    end
  end

  defp find_best_match([], _search, best), do: best
  defp find_best_match([index|indexes], search, best) do
    case match_index(index, search) do
      {_, [_,_]}  -> index   # Matching on both hash + sort makes this the best index (as well as we can tell)
      {_, [hash]} ->
        case search do
          # If we're only querying on a single field and we find a matching hash-only index, that's the index to use, no need to check others.
          [{field, {_, _}}] -> if field == hash, do: index, else: find_best_match(indexes, search, best)
          _                 -> find_best_match(indexes, search, index)   # we have a candidate for best match, though it's a hash key only. Look for better.
        end
      :not_found  ->
        case match_index_hash_part(index, search) do
          :not_found    -> find_best_match(indexes, search, best)    # haven't found anything good, keep looking, retain our previous best match.
          index_partial ->
            # If the current best is a hash-only index (formatted like {"idx_name", ["hash"]}), always choose it over a partial secondary.
            # Note that this default behavior would cause a hash-only key to be selected over a partial with a different hash
            # in cases where a query might be ambiguous - for example, a query on a user's first_name and last_name where
            # a hash-only index exists on first_name and a composite index exists on last_name_something_else.
            case best do
              {_, [_]} -> find_best_match(indexes, search, best)
              _        -> find_best_match(indexes, search, index_partial)
            end
        end
    end
  end


  # The parameter, 'search', is a map: %{field_name::string => {value::string, operator::atom}}
  defp match_index(index, search) do
    case index do
      {_, [hash, range]} ->
        # Part of the query could be a nil filter on an indexed attribute;
        # in that case, we need a check in addition to the key, so we check the operator.
        # Also, the hash part can only accept an :== operator.
        hash_key = deep_find_key(search, hash)
        range_key = deep_find_key(search, range)

        if hash_key != nil and elem(hash_key, 1) in [:==, :in]
        and range_key != nil and elem(range_key, 1) != :is_nil,
        do: index, else: :not_found

      {_, [hash]} ->
        hash_key = deep_find_key(search, hash)
        if hash_key != nil and elem(hash_key, 1) in [:==, :in], 
        do: index, else: :not_found
    end
  end

  @spec match_index_hash_part({:primary, [key]}, search) :: {:primary_partial, [key]} | :not_found
  @spec match_index_hash_part({String.t, [key]}, search) :: {:secondary_partial, String.t, [key]} | :not_found
  defp match_index_hash_part(index, search) do
    case index do
      {:primary, [hash, _range]} ->
        hash_key = deep_find_key(search, hash)
        if hash_key != nil and elem(hash_key, 1) in [:==, :in],
        do: {:primary_partial, [hash]}, else: :not_found

      {index_name, [hash, _range]} ->
        hash_key = deep_find_key(search, hash)
        if hash_key != nil and elem(hash_key, 1) in [:==, :in],
        do: {:secondary_partial, index_name, [hash]}, else: :not_found

      _ -> :not_found
    end
  end

  # TODO: multiple use of deep_find_key could be avoided by using the recursion in the main module to provide a set of indexed attributes in addition to the nested logical clauses.
  @spec deep_find_key(search, key) :: nil | {term, query_op}
  defp deep_find_key([], _), do: nil
  defp deep_find_key([clause | clauses], key) do
    case clause do
      {field, {val, op}} when not(field in [:and, :or]) ->
        if field == key, do: {val, op}, else: deep_find_key(clauses, key)
      {logical_op, deeper_clauses} when logical_op in [:and, :or] ->
        found = deep_find_key(deeper_clauses, key)
        if found != nil, do: found, else: deep_find_key(clauses, key)
    end
  end

  @doc """
  Formats the recursive option according to whether the query is a DynamoDB scan or query. (The adapter defaults to recursive fetch in case of the latter but not the former)
  """
  def parse_recursive_option(scan_or_query, opts) do
    case opts[:page_limit] do
      page_limit when (is_integer page_limit) and page_limit > 0 ->
        page_limit

      page_limit when (is_integer page_limit) and page_limit < 1 ->
        raise ArgumentError, message: "#{inspect __MODULE__}.parse_recursive_option/2 error: :page_limit option must be greater than 0."

      _ when scan_or_query == :scan ->
        # scan defaults to no recursion, opts[:recursive] must equal true to enable it
        opts[:recursive] == true

      _ when scan_or_query == :query ->
        # query defaults to recursion, opts[:recursive] must equal false to disable it
        opts[:recursive] != false
    end
  end


  # scan
  defp maybe_scan(table, [], opts) do
    scan_enabled = opts[:scan] == true || Application.get_env(:ecto_adapters_dynamodb, :scan_all) == true || Enum.member?(Application.get_env(:ecto_adapters_dynamodb, :scan_tables), table)

    cond do
      # TODO: we could use the cached scan and apply the search filters
      # ourselves when they are provided.
      Enum.member?(Application.get_env(:ecto_adapters_dynamodb, :cached_tables), table) and opts[:scan] != true ->
        Ecto.Adapters.DynamoDB.Cache.scan!(table)

      scan_enabled ->
        limit_option = opts[:limit] || Application.get_env(:ecto_adapters_dynamodb, :scan_limit)
        scan_limit = if is_integer(limit_option), do: [limit: limit_option], else: []
        updated_opts = Keyword.drop(opts, [:recursive, :limit, :scan]) ++ scan_limit

        fetch_recursive(&ExAws.Dynamo.scan/2, table, updated_opts, parse_recursive_option(:scan, opts), %{})

      true ->
        maybe_scan_error(table)
    end
  end

  defp maybe_scan(table, search, opts) do
    scan_enabled = opts[:scan] == true || Application.get_env(:ecto_adapters_dynamodb, :scan_all) == true || Enum.member?(Application.get_env(:ecto_adapters_dynamodb, :scan_tables), table)

    limit_option = opts[:limit] || Application.get_env(:ecto_adapters_dynamodb, :scan_limit)
    scan_limit = if is_integer(limit_option), do: [limit: limit_option], else: []
    updated_opts = Keyword.drop(opts, [:recursive, :limit, :scan]) ++ scan_limit

    if scan_enabled do
      {filter_expression_tuple, expression_attribute_names, expression_attribute_values} = construct_filter_expression(search, [])
        
      expressions = [
        expression_attribute_names: expression_attribute_names,
        expression_attribute_values: expression_attribute_values
      ] ++ updated_opts ++ filter_expression_tuple

      fetch_recursive(&ExAws.Dynamo.scan/2, table, expressions, parse_recursive_option(:scan, opts), %{})
    else
      maybe_scan_error(table)
    end
  end

  @spec maybe_scan_error(table_name) :: no_return
  defp maybe_scan_error(table) do
    raise ArgumentError, message: "#{inspect __MODULE__}.maybe_scan/3 error: :scan option or configuration have not been specified, and could not confirm the table, #{inspect table}, as listed for scan or caching in the application's configuration. Please see README file for details."
  end

  @typep fetch_func :: (table_name, keyword -> ExAws.Operation.JSON.t)
  @spec fetch_recursive(fetch_func, table_name, keyword, boolean | number, map) :: dynamo_response
  defp fetch_recursive(func, table, expressions, recursive, result) do
    updated_expressions = if recursive == true, do: Keyword.delete(expressions, :limit), else: expressions
    fetch_result = func.(table, updated_expressions) |> ExAws.request!
    # recursive can be a boolean or a page limit
    updated_recursive = update_recursive_option(recursive)

    if fetch_result["LastEvaluatedKey"] != nil and updated_recursive.continue do
      fetch_recursive(
        func,
        table,
        updated_expressions ++ [exclusive_start_key: fetch_result["LastEvaluatedKey"]],
        updated_recursive.new_value,
        combine_results(result, fetch_result)
      )
    else
      combine_results(result, fetch_result)
    end
  end

  @doc """
  Updates the recursive option during a recursive fetch, according to whether the option is a boolean or an integer (as in the case of page_limit)
  """
  def update_recursive_option(r) when (is_boolean r), do: %{continue: r,     new_value: r}
  def update_recursive_option(r) when (is_integer r), do: %{continue: r > 1, new_value: r - 1}

  @spec combine_results(map, map) :: map
  defp combine_results(result, scan_result) do
    if result == %{} do
      scan_result
    else
      %{"Count" => result_count, "Items" => result_items, "ScannedCount" => result_scanned_count} = result
      %{"Count" => scanned_count, "Items" => scanned_items, "ScannedCount" => scanned_scanned_count} = scan_result

      %{"Count" => result_count + scanned_count,
        "Items" => result_items ++ scanned_items,
        "ScannedCount" => result_scanned_count + scanned_scanned_count}
    end
  end

end