ecto_adapters_dynamodb

defmodule Ecto.Adapters.DynamoDB do
  @moduledoc """
  Ecto adapter for Amazon DynamoDB

  Currently for a fairly limited subset of Ecto, enough for basic operations.
  """

  #NOTE: in ecto, Repo.get[!] ends up calling: 
    #-> querable.get
    #-> queryable.one
    #-> queryable.all
    #-> queryable.execute
    #-> adapter.execute (possibly prepare somewhere in their too? trace.)


  @behaviour Ecto.Adapter
  #@behaviour Ecto.Adapter.Storage
  #@behaviour Ecto.Adapter.Migration

  defmacro __before_compile__(_env) do
    # Nothing to see here, yet...

  end

  use Bitwise, only_operators: true

  alias ExAws.Dynamo
  alias Ecto.Query.BooleanExpr

  # I don't think this is necessary: Probably under child_spec and ensure_all_started
  def start_link(repo, opts) do
    ecto_dynamo_log(:debug, "start_link repo: #{inspect repo} opts: #{inspect opts}")
    Agent.start_link fn -> [] end
  end


  ## Adapter behaviour - defined in lib/ecto/adapter.ex (in the ecto github repository)

  @doc """
  Returns the childspec that starts the adapter process.
  """
  def child_spec(repo, opts) do
    # TODO: need something here...
    # * Pull dynamo db connection options from config
    # * Start dynamo connector/aws libraries
    # we'll return our own start_link for now, but I don't think we actually need
    # an app here, we only need to ensure that our dependencies such as aws libs are started.
    # 

   [:debug_requests, :access_key_id, :secret_access_key, :region, :dynamodb] |> Enum.map(fn key ->
     if opts[key] != nil, do: Application.put_env(:ex_aws, key, opts[key])
   end)

    import Supervisor.Spec
    child_spec = worker(__MODULE__, [repo, opts])
    ecto_dynamo_log(:debug, "child spec3. REPO: #{inspect repo}\n CHILD_SPEC: #{inspect child_spec}\nOPTS: #{inspect opts}")
    child_spec
  end


  @doc """
  Ensure all applications necessary to run the adapter are started.
  """
  def ensure_all_started(repo, type) do
    ecto_dynamo_log(:debug, "ensure all started: type: #{inspect type} #{inspect repo}")
    with {:ok, _} = Application.ensure_all_started(:ecto_adapters_dynamodb)
    do
      {:ok, [repo]}
    end
  end

  def supports_ddl_transaction?, do: false

  def execute_ddl(repo, command, options) do
    Ecto.Adapters.DynamoDB.Migration.execute_ddl(repo, command, options)
  end


# moved to transaction.ex in ecto 2.1.4
#  def in_transaction?(_repo), do: false
#
#  def rollback(_repo, _value), do:
#    raise BadFunctionError, message: "#{inspect __MODULE__} does not support transactions."


  @doc """
  Called to autogenerate a value for id/embed_id/binary_id.

  Returns the autogenerated value, or nil if it must be
  autogenerated inside the storage or raise if not supported.

  For the Ecto type, `:id`, the adapter autogenerates a 128-bit integer

  For the Ecto type, `:embed_id`, the adapter autogenerates a string, using `Ecto.UUID.generate()`
  
  For the Ecto type, `:binary_id`, the adapter autogenerates a string, using `Ecto.UUID.generate()`
  """

  @max_id ((1 <<< 128) - 1) # biggest possible int in 128 bits
  def autogenerate(:id), do: Enum.random(1..@max_id)
  def autogenerate(:embed_id), do: Ecto.UUID.generate()
  def autogenerate(:binary_id), do: Ecto.UUID.generate()

  @doc """
  Returns the loaders for a given type.

  Rather than use the Ecto adapter loaders callback, the adapter builds on ExAws' decoding functionality, please see ExAws's `ExAws.Dynamo.decode_item` and the private function, `custom_decode`, in this module, which at this time only loads :utc_datetime and :naive_datetime.

  """
  def loaders(_primitive, type), do: [type]



  @doc """
  Returns the dumpers for a given type.

  We rely on ExAws encoding functionality during insertion and update to properly format types for DynamoDB. Please see ExAws `ExAws.Dynamo.update_item` and `ExAws.Dynamo.put_item` for specifics. Currently, we only modify :utc_datetime and :naive_datetime, appending the UTC offset, "Z", to the datetime string before passing to ExAws.
  """
  def dumpers(:utc_datetime, datetime), do: [datetime, &to_iso_string/1]
  def dumpers(:naive_datetime, datetime), do: [datetime, &to_iso_string/1]
  def dumpers(_primitive, type), do: [type]

  # Add UTC offset
  # We are adding the offset here also for the :naive_datetime, this
  # assumes we are getting a UTC date (which does correspond with the
  # timestamps() macro but not necessarily with :naive_datetime in general)
  defp to_iso_string(datetime) do
    {:ok, (datetime |> Ecto.DateTime.cast! |> Ecto.DateTime.to_iso8601) <> "Z"}
  end


  @doc """
  Commands invoked to prepare a query for `all`, `update_all` and `delete_all`.

  The returned result is given to `execute/6`.
  """
  #@callback prepare(atom :: :all | :update_all | :delete_all, query :: Ecto.Query.t) ::
  #          {:cache, prepared} | {:nocache, prepared}
  def prepare(:all, query) do
    # 'preparing' is more a SQL concept - Do we really need to do anything here or just pass the params through?
    ecto_dynamo_log(:debug, "PREPARE:::")
    ecto_dynamo_log(:debug, inspect(query, structs: false))
    {:nocache, {:all, query}}
  end
  

  def prepare(:update_all, query) do
    ecto_dynamo_log(:debug, "PREPARE::UPDATE_ALL:::")
    ecto_dynamo_log(:debug, inspect(query, structs: false))
    {:nocache, {:update_all, query}}
  end
  # do: {:cache, {System.unique_integer([:positive]), @conn.update_all(query)}}


  def prepare(:delete_all, query) do
    ecto_dynamo_log(:debug, "PREPARE::DELETE_ALL:::")
    ecto_dynamo_log(:debug, inspect(query, structs: false))
    {:nocache, {:delete_all, query}}
  end
  # do: {:cache, {System.unique_integer([:positive]), @conn.delete_all(query)}}



  @doc """
  Executes a previously prepared query.

  It must return a tuple containing the number of entries and
  the result set as a list of lists. The result set may also be
  `nil` if a particular operation does not support them.

  The `meta` field is a map containing some of the fields found
  in the `Ecto.Query` struct.

  It receives a process function that should be invoked for each
  selected field in the query result in order to convert them to the
  expected Ecto type. The `process` function will be nil if no
  result set is expected from the query.
  """
  #@callback execute(repo, query_meta, query, params :: list(), process | nil, options) :: result when
  #          result: {integer, [[term]] | nil} | no_return,
  #          query: {:nocache, prepared} |
  #                 {:cached, (prepared -> :ok), cached} |
  #                 {:cache, (cached -> :ok), prepared}
  def execute(repo, meta, {:nocache, {func, prepared}}, params, process, opts) do
    ecto_dynamo_log(:debug, "EXECUTE... EXECUTING!=============================")
    ecto_dynamo_log(:debug, "REPO::: #{inspect repo, structs: false}")
    ecto_dynamo_log(:debug, "META::: #{inspect meta, structs: false}")
    ecto_dynamo_log(:debug, "PREPARED::: #{inspect prepared, structs: false}")
    ecto_dynamo_log(:debug, "PARAMS::: #{inspect params, structs: false}")
    ecto_dynamo_log(:debug, "PROCESS::: #{inspect process, structs: false}")
    ecto_dynamo_log(:debug, "OPTS::: #{inspect opts, structs: false}")

    {table, model} = prepared.from
    validate_where_clauses!(prepared)
    lookup_fields = extract_lookup_fields(prepared.wheres, params, [])

    limit_option = opts[:scan_limit]
    scan_limit = if is_integer(limit_option), do: [limit: limit_option], else: []

    # Ecto migration does not know to specify 'scan: true' to retrieve the persisted migration versions
    # from line 34, file "deps/ecto/lib/ecto/migration/schema_migration.ex"
    migration_source = Keyword.get(repo.config, :migration_source, "schema_migrations")
    updated_opts = if table == migration_source do
      ecto_dynamo_log(:debug, "Table name corresponds with migration source: #{inspect migration_source}. Setting options for recursive scan.")
      Keyword.drop(opts, [:timeout, :log]) ++ [recursive: true]
    else
      Keyword.drop(opts, [:scan_limit, :limit]) ++ scan_limit
    end

    ecto_dynamo_log(:debug, "table = #{inspect table}")
    ecto_dynamo_log(:debug, "lookup_fields: #{inspect lookup_fields}")
    ecto_dynamo_log(:debug, "scan_limit: #{inspect scan_limit}")

    case func do
      :delete_all ->
        ecto_dynamo_log(:info, "#{inspect __MODULE__}.execute: delete_all")
        ecto_dynamo_log(:info, "Table: #{inspect table}; Lookup fields: #{inspect lookup_fields}; Options: #{inspect updated_opts}")
        delete_all(table, lookup_fields, updated_opts)

      :update_all  -> 
        ecto_dynamo_log(:info, "#{inspect __MODULE__}.execute: update_all")
        ecto_dynamo_log(:info, "Table: #{inspect table}; Lookup fields: #{inspect lookup_fields}; Options: #{inspect updated_opts}")

        update_all(table, lookup_fields, updated_opts, prepared.updates, params)

      :all ->
        ecto_dynamo_log(:info, "#{inspect __MODULE__}.execute")
        ecto_dynamo_log(:info, "Table: #{inspect table}; Lookup fields: #{inspect lookup_fields}; options: #{inspect updated_opts}")
        result = Ecto.Adapters.DynamoDB.Query.get_item(table, lookup_fields, updated_opts)
        ecto_dynamo_log(:debug, "result = #{inspect result}")

        if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], extract_query_info(result))

        if result == %{} do
          # Empty map means "not found"
          {0, []}
        else
          cond do
            !result["Count"] and !result["Responses"] -> 
              decoded = result |> Dynamo.decode_item(as: model) |> custom_decode(model, prepared.select)
              {1, [decoded]}

            true ->
              # batch_get_item returns "Responses" rather than "Items"
              results_to_decode = if result["Items"], do: result["Items"], else: result["Responses"][table]

              decoded = Enum.map(results_to_decode, fn(item) -> 
                Dynamo.decode_item(%{"Item" => item}, as: model) |> custom_decode(model, prepared.select)
              end)

              {length(decoded), decoded}
          end
        end
    end
  end

  # delete_all allows for the recursive option, scanning through multiple pages
  defp delete_all(table, lookup_fields, opts) do
    # select only the key
    {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table)
    scan_or_query = Ecto.Adapters.DynamoDB.Query.scan_or_query?(table, lookup_fields)
    recursive = Ecto.Adapters.DynamoDB.Query.parse_recursive_option(scan_or_query, opts)
    updated_opts = prepare_recursive_opts(opts ++ [projection_expression: Enum.join(key_list, ", ")])

    delete_all_recursive(table, lookup_fields, updated_opts, recursive, %{}, 0)
  end

  defp delete_all_recursive(table, lookup_fields, opts, recursive, query_info, total_processed) do
    # query the table for which records to delete
    fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, lookup_fields, opts)

    ecto_dynamo_log(:debug, "fetch_result = #{inspect fetch_result}")

    items = case fetch_result do
      %{"Items" => fetch_items}   -> fetch_items
      %{"Item" => item}           -> [item]
      %{"Responses" => table_map} -> table_map[table]
      _                           -> []
    end

    prepared_data = for key_list <- Enum.map(items, &Map.to_list/1) do
      key_map = for {key, val_map} <- key_list, into: %{}, do: {key, Dynamo.Decoder.decode(val_map)}
      [delete_request: [key: key_map]]
    end

    unprocessed_items = if prepared_data != [] do
      batch_delete(table, prepared_data)
    else
      %{}
    end

    num_processed =
      length(prepared_data) - if !unprocessed_items[table], do: 0, else: length(unprocessed_items[table])

    updated_query_info = Enum.reduce(fetch_result, query_info, fn({key, val}, acc) ->
      case key do
        "Count" -> Map.update(acc, key, val, fn x -> x + val end)
        "ScannedCount" -> Map.update(acc, key, val, fn x -> x + val end)
        "LastEvaluatedKey" -> Map.update(acc, key, val, fn _ -> fetch_result["LastEvaluatedKey"] end)
        _ -> acc
      end
    end) |> Map.update("UnprocessedItems", unprocessed_items, fn map -> if map == %{}, do: %{}, else: %{table => map[table] ++ unprocessed_items[table]} end)

    updated_recursive = Ecto.Adapters.DynamoDB.Query.update_recursive_option(recursive)

    if fetch_result["LastEvaluatedKey"] != nil and updated_recursive.continue do
        opts_with_offset = opts ++ [exclusive_start_key: fetch_result["LastEvaluatedKey"]]
        delete_all_recursive(table, lookup_fields, opts_with_offset, updated_recursive.new_value, updated_query_info, total_processed + num_processed)
    else
      # We're not retrying unprocessed items yet, but we are providing the relevant info in the QueryInfo agent if :query_info_key is supplied
      if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], updated_query_info)
      {num_processed + total_processed, nil}
    end
  end

  # returns unprocessed_items
  defp batch_delete(table, prepared_data) do
    batch_write_attempt = Dynamo.batch_write_item(%{table => prepared_data}) |> ExAws.request |> handle_error!(%{table: table, records: []})

    batch_write_attempt["UnprocessedItems"]
  end


  defp update_all(table, lookup_fields, opts, updates, params) do
    scan_or_query = Ecto.Adapters.DynamoDB.Query.scan_or_query?(table, lookup_fields)
    recursive = Ecto.Adapters.DynamoDB.Query.parse_recursive_option(scan_or_query, opts)

    key_list = Ecto.Adapters.DynamoDB.Info.primary_key!(table)
    ecto_dynamo_log(:debug, "key_list: #{inspect key_list}")

    # The remove statement must be constructed after finding pull-indexes, but it
    # also includes possibly removing nil fields, and since we have one handler for
    # both set and remove, we call it during the batch update process
    {update_expression, update_fields_sans_set_remove, set_remove_fields} = construct_update_expression(updates, params, opts)
    ecto_dynamo_log(:info, "update fields: #{inspect update_fields_sans_set_remove} set and remove fields: #{inspect set_remove_fields}")
    attribute_names = construct_expression_attribute_names(update_fields_sans_set_remove)
    attribute_values = construct_expression_attribute_values(update_fields_sans_set_remove, opts)

    base_update_options = [expression_attribute_names: attribute_names,
                           update_expression: update_expression,
                           return_values: :all_new]

    updated_opts = prepare_recursive_opts(opts)
    update_options = maybe_add_attribute_values(base_update_options, attribute_values)

    pull_actions_without_index =
      Keyword.keys(set_remove_fields[:pull]) 
      |> Enum.any?(fn x -> !Enum.member?(Keyword.keys(maybe_list(opts[:pull_indexes])), x) end)

    {new_update_options, new_set_remove_fields} =
      if pull_actions_without_index do
        {update_options, set_remove_fields}
      else
        merged_pull_indexes = Keyword.merge(set_remove_fields[:pull], maybe_list(opts[:pull_indexes]))
        opts_with_pull_indexes = Keyword.update(opts, :pull_indexes, merged_pull_indexes, fn _ -> merged_pull_indexes end)
        {update_batch_update_options(update_options, set_remove_fields, opts_with_pull_indexes), []}
      end

    update_all_recursive(table, lookup_fields, updated_opts, new_update_options, key_list, new_set_remove_fields, recursive, %{}, 0)
  end

  defp update_all_recursive(table, lookup_fields, opts, update_options, key_list, set_remove_fields, recursive, query_info, total_updated) do
    fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, lookup_fields, opts)
    ecto_dynamo_log(:debug, "fetch_result: #{inspect fetch_result}")

    updated_query_info = case fetch_result do
      %{"Count" => last_count, "ScannedCount" => last_scanned_count} -> 
        %{"Count" => last_count + Map.get(query_info, "Count", 0),
          "ScannedCount" => last_scanned_count + Map.get(query_info, "ScannedCount", 0),
          "LastEvaluatedKey" => Map.get(fetch_result, "LastEvaluatedKey")}

      _ -> query_info
    end

    items = case fetch_result do
      %{"Items" => fetch_items}   -> fetch_items
      %{"Item" => item}           -> [item]
      %{"Responses" => table_map} -> table_map[table]
      _                           -> []
    end

    num_updated = if items != [] do
      batch_update(table, items, key_list, update_options, set_remove_fields, opts)
    else
      0
    end

    updated_recursive = Ecto.Adapters.DynamoDB.Query.update_recursive_option(recursive)

    if fetch_result["LastEvaluatedKey"] != nil and updated_recursive.continue do
        opts_with_offset = opts ++ [exclusive_start_key: fetch_result["LastEvaluatedKey"]]
        update_all_recursive(table, lookup_fields, opts_with_offset, update_options, key_list, set_remove_fields, updated_recursive.new_value, updated_query_info, total_updated + num_updated)
    else
      if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], updated_query_info)
      {total_updated + num_updated, []}
    end
  end

  defp batch_update(table, items, key_list, update_options, set_remove_fields, opts) do
    Enum.reduce(items, 0, fn(result_to_update, acc) ->
      filters = get_key_values_dynamo_map(result_to_update, key_list)

      # we only update this on a case-by-case basis if pull actions
      # without specific indexes are specified
      options_with_set_and_remove = case set_remove_fields do
        [] -> update_options
        _  ->
          pull_fields_with_indexes =
            Enum.map(set_remove_fields[:pull], fn {field_atom, val} -> 
              list = result_to_update[to_string(field_atom)]
              {field_atom, find_all_indexes_in_dynamodb_list(list, val)}
            end)
          merged_pull_indexes = Keyword.merge(pull_fields_with_indexes, maybe_list(opts[:pull_indexes]))
          opts_with_pull_indexes = Keyword.update(opts, :pull_indexes, merged_pull_indexes, fn _ -> merged_pull_indexes end)

          update_batch_update_options(update_options, set_remove_fields, opts_with_pull_indexes)
      end

      # 'options_with_set_and_remove' might not have the key, ':expression_attribute_values',
      # when there are only removal statements.
      record = if options_with_set_and_remove[:expression_attribute_values],
               do: [options_with_set_and_remove[:expression_attribute_values] |> Enum.into(%{})],
               else: []

      if options_with_set_and_remove[:update_expression] |> String.trim != "" do
        Dynamo.update_item(table, filters, options_with_set_and_remove) |> ExAws.request |> handle_error!(%{table: table, records: record ++ []})
        acc + 1
      else
        acc
      end
    end)
  end

  defp update_batch_update_options(update_options, set_remove_fields, opts) do
    attribute_names = construct_expression_attribute_names(Keyword.values(set_remove_fields) |> List.flatten)
    set_and_push_fields = maybe_list(set_remove_fields[:set]) ++ maybe_list(set_remove_fields[:push])
    opts_with_push = opts ++ Keyword.take(set_remove_fields, [:push])
    attribute_values = construct_expression_attribute_values(set_and_push_fields, opts_with_push)
    set_statement = construct_set_statement(set_remove_fields[:set], opts_with_push)
    opts_for_construct_remove = Keyword.take(set_remove_fields, [:pull]) ++ Keyword.take(opts, [:pull_indexes, :remove_nil_fields])
    remove_statement = construct_remove_statement(set_remove_fields[:set], opts_for_construct_remove)

    base_update_options =
      [expression_attribute_names: Map.merge(attribute_names, update_options[:expression_attribute_names]),
      update_expression: set_statement <> " " <> remove_statement <> " " <> update_options[:update_expression] |> String.trim,
      return_values: :all_new]

    maybe_add_attribute_values(base_update_options, attribute_values ++ maybe_list(update_options[:expression_attribute_values]))
  end

  # find indexes to remove for update :pull action
  defp find_all_indexes_in_dynamodb_list(dynamodb_list, target) do
    Dynamo.Decoder.decode(dynamodb_list)
    |> Enum.with_index()
    |> Enum.filter(fn {x, _} -> x == target end)
    |> Enum.map(fn {_, i} -> i end)
  end


  # During delete_all's and update_all's recursive
  # procedure, we want to keep the recursion in
  # the top-level, between actions, rather than
  # load all the results into memory and then act;
  # so we disable the recursion on get_item
  defp prepare_recursive_opts(opts) do
    opts |> Keyword.delete(:page_limit) |> Keyword.update(:recursive, false, fn _ -> false end)
  end


  @doc """
  Inserts a single new struct in the data store.

  ## Autogenerate

  The primary key will be automatically included in `returning` if the
  field has type `:id` or `:binary_id` and no value was set by the
  developer or none was autogenerated by the adapter.
  """
  #@callback insert(repo, schema_meta, fields, on_conflict, returning, options) ::
  #                  {:ok, fields} | {:invalid, constraints} | no_return
  #  def insert(_,_,_,_,_) do
  def insert(repo, schema_meta, fields, on_conflict, returning, opts) do
    ecto_dynamo_log(:debug, "INSERT::\n\trepo: #{inspect repo}")
    ecto_dynamo_log(:debug, "\tschema_meta: #{inspect schema_meta}")
    ecto_dynamo_log(:debug, "\tfields: #{inspect fields}")
    ecto_dynamo_log(:debug, "\ton_conflict: #{inspect on_conflict}")
    ecto_dynamo_log(:debug, "\treturning: #{inspect returning}")
    ecto_dynamo_log(:debug, "\topts: #{inspect opts}")

    insert_nil_field_option = Keyword.get(opts, :insert_nil_fields, true)
    do_not_insert_nil_fields = insert_nil_field_option == false || Application.get_env(:ecto_adapters_dynamodb, :insert_nil_fields) == false

    {_, table} = schema_meta.source 
    model = schema_meta.schema
    fields_map = Enum.into(fields, %{})
    record = if do_not_insert_nil_fields, do: fields_map, else: build_record_map(model, fields_map)

    ecto_dynamo_log(:info, "#{inspect __MODULE__}.insert")
    ecto_dynamo_log(:info, "Table: #{inspect table}; Record: #{inspect record}")

    {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table)
    hash_key = hd(key_list)

    on_conflict_action = elem(on_conflict, 0)

    options = case on_conflict_action do
      :replace_all -> []
      _ -> 
        attribute_names = for k <- key_list, into: %{}, do: {"##{k}", k}
        conditions = for k <- key_list, do: "attribute_not_exists(##{k})"
        condition_expression = Enum.join(conditions, " and ")
        [expression_attribute_names: attribute_names,
         condition_expression: condition_expression]
    end

    case Dynamo.put_item(table, record, options) |> ExAws.request |> handle_error!(%{table: table, records: [record]}) do
      {:error, "ConditionalCheckFailedException"} ->
        case on_conflict_action do
          :nothing -> {:ok, (for key <- key_list, do: {String.to_atom(key), nil})}
          :raise   ->
            # This constraint name yields the correct behavior in the case the user
            # has specified a unique constraint on the primary key in their schema:
            constraint_name = "#{table}_#{hash_key}_index"
            {:invalid, [unique: constraint_name]}
        end

      %{} ->
        {:ok, []}
    end
  end


  def insert_all(repo, schema_meta, field_list, fields, on_conflict, returning, opts) do
    ecto_dynamo_log(:debug, "INSERT ALL::\n\trepo: #{inspect repo}")
    ecto_dynamo_log(:debug, "\tschema_meta: #{inspect schema_meta}")
    ecto_dynamo_log(:debug, "\tfield_list: #{inspect field_list}")
    ecto_dynamo_log(:debug, "\tfields: #{inspect fields}")
    ecto_dynamo_log(:debug, "\ton_conflict: #{inspect on_conflict}")
    ecto_dynamo_log(:debug, "\treturning: #{inspect returning}")
    ecto_dynamo_log(:debug, "\topts: #{inspect opts}")

    insert_nil_field_option = Keyword.get(opts, :insert_nil_fields, true)
    do_not_insert_nil_fields = insert_nil_field_option == false || Application.get_env(:ecto_adapters_dynamodb, :insert_nil_fields) == false

    {_, table} = schema_meta.source
    model = schema_meta.schema

    prepared_fields = Enum.map(fields, fn(field_set) ->
      mapped_fields = Enum.into(field_set, %{})
      record = if do_not_insert_nil_fields, do: mapped_fields, else: build_record_map(model, mapped_fields)

      [put_request: [item: record]]
    end)

    records = Enum.map(prepared_fields, fn [put_request: [item: record]] -> record end)

    ecto_dynamo_log(:info, "#{inspect __MODULE__}.insert_all")
    ecto_dynamo_log(:info, "Table: #{inspect table}; Records: #{inspect records}")

    batch_write_attempt = Dynamo.batch_write_item(%{table => prepared_fields}) |> ExAws.request |> handle_error!(%{table: table, records: records})

    ecto_dynamo_log(:debug, "Batch-write result: #{inspect batch_write_attempt}")

    # We're not retrying unprocessed items yet, but we are providing the relevant info in the QueryInfo agent if :query_info_key is supplied
    if opts[:query_info_key], do: Ecto.Adapters.DynamoDB.QueryInfo.put(opts[:query_info_key], extract_query_info(batch_write_attempt))

    if batch_write_attempt["UnprocessedItems"] == %{} do
      {length(records), nil}
    else
      {length(records) - length(batch_write_attempt["UnprocessedItems"][table]), nil}
    end
  end

  defp build_record_map(model, fields_to_insert) do
    # Ecto does not convert empty strings to nil before passing them
    # to Repo.insert_all, and ExAws will remove empty strings (as well as empty lists)
    # when building the insertion query but not nil values. We don't mind the removal
    # of empty lists since those cannot be inserted to indexed fields, but we'd like to
    # catch the removal of fields with empty strings by ExAws to support our option, :remove_nil_fields,
    # so we convert these to nil.
    empty_strings_to_nil = fields_to_insert 
                         |> Enum.map(fn {field, val} -> {field, (if val == "", do: nil, else: val)} end)
                         |> Enum.into(%{})
    model.__struct__ |> Map.delete(:__meta__) |> Map.from_struct |> Map.merge(empty_strings_to_nil)
  end


  # In testing, 'filters' contained only the primary key and value 
  # TODO: handle cases of more than one tuple in 'filters'?
  def delete(repo, schema_meta, filters, opts) do
    ecto_dynamo_log(:debug, "DELETE::\n\trepo: #{inspect repo}")
    ecto_dynamo_log(:debug, "\tschema_meta: #{inspect schema_meta}")
    ecto_dynamo_log(:debug, "\tfilters: #{inspect filters}")
    ecto_dynamo_log(:debug, "\toptions: #{inspect opts}")

    {_, table} = schema_meta.source

    # We offer the :range_key option for tables with composite primary key
    # since Ecto will not provide the range_key value needed for the query.
    # If :range_key is not provided, check if the table has a composite
    # primary key and query for all the key values
    updated_filters = case opts[:range_key] do
      nil -> 
        {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table)
        if (length key_list) > 1 do
          updated_opts = opts ++ [projection_expression: Enum.join(key_list, ", ")]
          filters_as_strings = for {field, val} <- filters, do: {Atom.to_string(field), {val, :==}}
          fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, filters_as_strings, updated_opts)
          items = case fetch_result do
            %{"Items" => fetch_items} -> fetch_items
            %{"Item" => item}         -> [item]
            _                         -> []
          end

          if items == [], do: raise "__MODULE__.update error: no results found for record: #{inspect filters}"
          if (length items) > 1, do: raise "__MODULE__.update error: more than one result found for record: #{inspect filters}"
          
          for {field, key_map} <- Map.to_list(hd items) do
            [{_field_type, val}] = Map.to_list(key_map)
            {field, val}
          end
         else
          filters
         end

      range_key ->
        [range_key | filters]
    end

    attribute_names = construct_expression_attribute_names(keys_to_atoms(filters))

    base_options = [expression_attribute_names: attribute_names]
    condition_expression = construct_condition_expression(filters)
    options = base_options ++ [condition_expression: condition_expression]

    # 'options' might not have the key, ':expression_attribute_values', when there are only removal statements
    record = if options[:expression_attribute_values], do: [options[:expression_attribute_values] |> Enum.into(%{})], else: []

    case Dynamo.delete_item(table, updated_filters, options) |> ExAws.request |> handle_error!(%{table: table, records: record ++ []}) do
      %{} -> {:ok, []}
      {:error, "ConditionalCheckFailedException"} -> {:error, :stale}
    end
  end


  def update(repo, schema_meta, fields, filters, returning, opts) do
    ecto_dynamo_log(:debug, "UPDATE::\n\trepo: #{inspect repo}")
    ecto_dynamo_log(:debug, "\tschema_meta: #{inspect schema_meta}")
    ecto_dynamo_log(:debug, "\tfields: #{inspect fields}")
    ecto_dynamo_log(:debug, "\tfilters: #{inspect filters}")
    ecto_dynamo_log(:debug, "\treturning: #{inspect returning}")
    ecto_dynamo_log(:debug, "\toptions: #{inspect opts}")

    {_, table} = schema_meta.source

    # We offer the :range_key option for tables with composite primary key
    # since Ecto will not provide the range_key value needed for the query.
    # If :range_key is not provided, check if the table has a composite
    # primary key and query for all the key values
    updated_filters = case opts[:range_key] do
      nil -> 
        {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(table)
        if (length key_list) > 1 do
          updated_opts = opts ++ [projection_expression: Enum.join(key_list, ", ")]
          filters_as_strings = for {field, val} <- filters, do: {Atom.to_string(field), {val, :==}}
          fetch_result = Ecto.Adapters.DynamoDB.Query.get_item(table, filters_as_strings, updated_opts)
          items = case fetch_result do
            %{"Items" => fetch_items} -> fetch_items
            %{"Item" => item}         -> [item]
            _                         -> []
          end

          if items == [], do: raise "__MODULE__.update error: no results found for record: #{inspect filters}"
          if (length items) > 1, do: raise "__MODULE__.update error: more than one result found for record: #{inspect filters}"
          
          for {field, key_map} <- Map.to_list(hd items) do
            [{_field_type, val}] = Map.to_list(key_map)
            {field, val}
          end
         else
          filters
         end

      range_key ->
        [range_key | filters]
    end

    update_expression = construct_update_expression(fields, opts)
    # add updated_filters to attribute_ names and values for condition_expression
    attribute_names = construct_expression_attribute_names(fields ++ keys_to_atoms(filters))
    attribute_values = construct_expression_attribute_values(fields, opts)

    base_options = [expression_attribute_names: attribute_names,
                    update_expression: update_expression]
    condition_expression = construct_condition_expression(filters)
    options = maybe_add_attribute_values(base_options, attribute_values)
           ++ [condition_expression: condition_expression]

    # 'options' might not have the key, ':expression_attribute_values', when there are only removal statements
    record = if options[:expression_attribute_values], do: [options[:expression_attribute_values] |> Enum.into(%{})], else: []

    case Dynamo.update_item(table, updated_filters, options) |> ExAws.request |> handle_error!(%{table: table, records: record ++ []}) do
      %{} -> {:ok, []}
      {:error, "ConditionalCheckFailedException"} -> {:error, :stale}
    end
  end

  defp keys_to_atoms(list),
  do: for {k, v} <- list, do: {maybe_string_to_atom(k), v}

  defp maybe_string_to_atom(s),
  do: if is_binary(s), do: String.to_atom(s), else: s

  defp construct_condition_expression([{field, _val}] = _filters),
  do: "attribute_exists(##{to_string(field)})"

  defp extract_query_info(result), do: result |> Map.take(["Count", "ScannedCount", "LastEvaluatedKey", "UnprocessedItems", "UnprocessedKeys"])


  # Used in update_all
  defp extract_update_params([], _action_atom, _params), do: []
  defp extract_update_params([%{expr: key_list}], action_atom, params) do
    case key_list[action_atom] do
      nil ->
        []
      action_list ->
        for s <- action_list do
          {field_atom, {:^, _, [idx]}} = s
          {field_atom, Enum.at(params,idx)}
        end
    end
  end

  defp extract_update_params([a], _action_atom, _params), do: error "#{inspect __MODULE__}.extract_update_params: Updates is either missing the :expr key or does not contain a struct or map: #{inspect a}"
  defp extract_update_params(unsupported, _action_atom, _params), do: error "#{inspect __MODULE__}.extract_update_params: unsupported parameter construction. #{inspect unsupported}"

  # Ecto does not support push pull for types other than array.
  # Therefore, we enable add and delete via opts
  defp extract_update_params(key_list, action_atom) do
    case key_list[action_atom] do
      nil         -> []
      action_list -> action_list
    end
  end

  # used in :update_all
  defp get_key_values_dynamo_map(dynamo_map, {:primary, keys}) do
    for k <- keys, do: {String.to_atom(k), Dynamo.Decoder.decode(dynamo_map[k])}
  end


  defp construct_expression_attribute_names(fields) do
    for {f, _} <- fields, into: %{}, do: {"##{Atom.to_string(f)}", Atom.to_string(f)}
  end

  defp construct_expression_attribute_values(fields, opts) do
    remove_rather_than_set_to_null = opts[:remove_nil_fields] || Application.get_env(:ecto_adapters_dynamodb, :remove_nil_fields_on_update) == true

    # If the value is nil and the :remove_nil_fields option is set, 
    # we're removing this attribute, not updating it, so filter out any such fields:

    if remove_rather_than_set_to_null do
      for {k, v} <- fields, !is_nil(v), do: {k, format_val(k, v, opts)}
    else
      for {k, v} <- fields, do: {k, format_nil(k, v, opts)}
    end |> Enum.filter(fn {x, _} -> not Keyword.has_key?(maybe_list(opts[:pull]), x) end)
  end

  defp maybe_list(l) when is_list(l), do: l
  defp maybe_list(_), do: []

  defp format_nil(_k, v, _opts) when is_nil(v), do: %{"NULL" => "true"}
  defp format_nil(k, v, opts), do: format_val(k, v, opts)

  defp format_val(k, v, opts) do
    case opts[:push][k] do
      nil -> v
      _   -> [v]
    end
  end

  # DynamoDB throws an error if we pass in an empty list for attribute values,
  # so we have to implement this stupid little helper function to avoid hurting
  # its feelings:
  defp maybe_add_attribute_values(options, []) do
    options
  end
  defp maybe_add_attribute_values(options, attribute_values) do
    [expression_attribute_values: attribute_values] ++ options
  end

  defp construct_update_expression(updates, params, opts) do
    to_set = extract_update_params(updates, :set, params)
    to_push = extract_update_params(updates, :push, params)
    to_pull = extract_update_params(updates, :pull, params)
    to_add = extract_update_params(opts, :add) ++ extract_update_params(updates, :inc, params)
    to_delete = extract_update_params(opts, :delete)

    {construct_add_statement(to_add, opts) <> " " <>
     construct_delete_statement(to_delete, opts) |> String.trim(),
     to_add ++ to_delete,
     [set: to_set, push: to_push, pull: to_pull]}
  end

  # The update callback supplies fields in the paramaters
  # whereas update_all includes a more complicated updates
  # structure
  defp construct_update_expression(fields, opts) do
    set_statement = construct_set_statement(fields, opts)
    rem_statement = construct_remove_statement(fields, opts)

    String.trim("#{set_statement} #{rem_statement}")
  end

  # fields::[{:field, val}]
  defp construct_set_statement(fields, opts) do
    remove_rather_than_set_to_null = opts[:remove_nil_fields] || Application.get_env(:ecto_adapters_dynamodb, :remove_nil_fields_on_update) == true

    set_clauses = for {key, val} <- fields, not (is_nil(val) and remove_rather_than_set_to_null) do
      key_str = Atom.to_string(key)
      "##{key_str}=:#{key_str}"
    end
    ++ case opts[:push] do
      nil       -> []
      push_list ->
        for {key, _val} <- push_list do
          key_str = Atom.to_string(key)
          if Enum.member?(maybe_list(opts[:prepend_to_list]), key),
          do: "##{key_str} = list_append(:#{key_str}, ##{key_str})",
          else: "##{key_str} = list_append(##{key_str}, :#{key_str})"
        end
    end

    case set_clauses do
      [] ->
        ""
      _ ->
        "SET " <> Enum.join(set_clauses, ", ")
    end
  end

  defp construct_remove_statement(fields, opts) do
    remove_rather_than_set_to_null = opts[:remove_nil_fields] || Application.get_env(:ecto_adapters_dynamodb, :remove_nil_fields_on_update) == true

    remove_clauses =
      if remove_rather_than_set_to_null do
        for {key, val} <- fields, is_nil(val), do: "##{Atom.to_string(key)}"
      else
        []
      end

    # Ecto :pull update can be emulated provided
    # we are given an index to remove in opts[:pull_indexes]
    ++ cond do
      !opts[:pull_indexes] or (Keyword.values(opts[:pull_indexes]) |> List.flatten) == [] ->
        []
      opts[:pull] == nil ->
        []
      true ->
        for {key, _val} <- opts[:pull] do
          key_str = Atom.to_string(key)
          Enum.map(opts[:pull_indexes][key], fn index -> "##{key_str}[#{index}]" end) |> Enum.join(", ")
        end
    end

    case remove_clauses do
      [] ->
        ""
      _ ->
        "REMOVE " <> Enum.join(remove_clauses, ", ")
    end
  end

  # fields::[{:field, val}]
  defp construct_add_statement(fields, _opts) do
    add_clauses = for {key, _val} <- fields do
      key_str = Atom.to_string(key)
      "##{key_str} :#{key_str}"
    end
    case add_clauses do
      [] ->
        ""
      _ ->
        "ADD " <> Enum.join(add_clauses, ", ")
    end
  end

  defp construct_delete_statement(fields, _opts) do
    delete_clauses = for {key, _val} <- fields do
      key_str = Atom.to_string(key)
      "##{key_str} :#{key_str}"
    end
    case delete_clauses do
      [] ->
        ""
      _ ->
        "DELETE " <> Enum.join(delete_clauses, ", ")
    end
  end

  defp validate_where_clauses!(query) do
    for w <- query.wheres do
      validate_where_clause! w
    end
  end
  defp validate_where_clause!(%BooleanExpr{expr: {op, _, _}}) when op in [:==, :<, :>, :<=, :>=, :in], do: :ok
  defp validate_where_clause!(%BooleanExpr{expr: {logical_op, _, _}}) when logical_op in [:and, :or], do: :ok
  defp validate_where_clause!(%BooleanExpr{expr: {:is_nil, _, _}}), do: :ok
  defp validate_where_clause!(%BooleanExpr{expr: {:fragment, _, _}}), do: :ok
  defp validate_where_clause!(unsupported), do: error "unsupported where clause: #{inspect unsupported}"

  # We are parsing a nested, recursive structure of the general type:
  # %{:logical_op, list_of_clauses} | %{:conditional_op, field_and_value}
  defp extract_lookup_fields([], _params, lookup_fields), do: lookup_fields
  defp extract_lookup_fields([query | queries], params, lookup_fields) do
    # A logical operator tuple does not always have a parent 'expr' key.
    maybe_extract_from_expr = case query do
      %BooleanExpr{expr: expr} -> expr
      # TODO: could there be other cases?
      _                        -> query
    end

    case maybe_extract_from_expr do 
      # A logical operator points to a list of conditionals
      {op, _, [left, right]} when op in [:==, :<, :>, :<=, :>=, :in] ->
        {field, value} = get_op_clause(left, right, params)
        updated_lookup_fields = 
          case List.keyfind(lookup_fields, field, 0) do
            # we assume the most ops we can apply to one field is two, otherwise this might throw an error
            {field, {old_val, old_op}} ->
              List.keyreplace(lookup_fields, field, 0, {field, {[value, old_val], [op, old_op]}})

            _ -> [{field, {value, op}} | lookup_fields]
          end
        extract_lookup_fields(queries, params, updated_lookup_fields)

      # Logical operator expressions have more than one op clause
      # We are matching queries of the type: 'from(p in Person, where: p.email == "g@email.com" and p.first_name == "George")'
      # But not of the type: 'from(p in Person, where: [email: "g@email.com", first_name: "George"])'
      #
      # A logical operator is a member of a list
      {logical_op, _, clauses} when logical_op in [:and, :or] ->
        deeper_lookup_fields = extract_lookup_fields(clauses, params, [])
        extract_lookup_fields(queries, params, [{logical_op, deeper_lookup_fields} | lookup_fields])

      {:fragment, _, raw_expr_mixed_list} ->
        parsed_fragment = parse_raw_expr_mixed_list(raw_expr_mixed_list, params)
        extract_lookup_fields(queries, params, [parsed_fragment | lookup_fields])
        
      # We perform a post-query is_nil filter on indexed fields and have DynamoDB filter
      # for nil non-indexed fields (although post-query nil-filters on (missing) indexed 
      # attributes could only find matches when the attributes are not the range part of 
      # a queried partition key (hash part) since those would not return the sought records).
      {:is_nil, _, [arg]} ->
        {{:., _, [_, field_name]}, _, _} = arg

        # We give the nil value a string, "null", since it will be mapped as a DynamoDB attribute_expression_value
        extract_lookup_fields(queries, params, [{to_string(field_name), {"null", :is_nil}} | lookup_fields])

      _ -> extract_lookup_fields(queries, params, lookup_fields)
    end
  end

  # Specific (as opposed to generalized) parsing for Ecto :fragment - the only use for it
  # so far is 'between' which is the only way to query 'between' on an indexed field since
  # those accept only single conditions.
  #
  # Example with values as strings: [raw: "", expr: {{:., [], [{:&, [], [0]}, :person_id]}, [], []}, raw: " between ", expr: "person:a", raw: " and ", expr: "person:f", raw: ""]
  #
  # Example with values as part of the string itself: [raw: "", expr: {{:., [], [{:&, [], [0]}, :person_id]}, [], []}, raw: " between person:a and person:f"]
  #
  # Example with values in params: [raw: "", expr: {{:., [], [{:&, [], [0]}, :person_id]}, [], []}, raw: " between ", expr: {:^, [], [0]}, raw: " and ", expr: {:^, [], [1]}, raw: ""]
  #
  defp parse_raw_expr_mixed_list(raw_expr_mixed_list, params) do
    # group the expression into fields, values, and operators,
    # only supporting the example with values in params
    case raw_expr_mixed_list do
      # between
      [raw: _, expr: {{:., [], [{:&, [], [0]}, field_atom]}, [], []}, raw: between_str, expr: {:^, [], [idx1]}, raw: and_str, expr: {:^, [], [idx2]}, raw: _] ->
        if not (Regex.match?(~r/^\s*between\s*and\s*$/i, between_str <> and_str)), do:
          parse_raw_expr_mixed_list_error(raw_expr_mixed_list)
        {to_string(field_atom), {[Enum.at(params, idx1), Enum.at(params, idx2)], :between}}

      # begins_with
      [raw: begins_with_str, expr: {{:., [], [{:&, [], [0]}, field_atom]}, [], []}, raw: comma_str, expr: {:^, [], [idx]}, raw: closing_parenthesis_str] ->
        if not (Regex.match?(~r/^\s*begins_with\(\s*,\s*\)\s*$/i, begins_with_str <> comma_str <> closing_parenthesis_str)), do:
          parse_raw_expr_mixed_list_error(raw_expr_mixed_list)
        {to_string(field_atom), {Enum.at(params, idx), :begins_with}}

      _ -> parse_raw_expr_mixed_list_error(raw_expr_mixed_list) 
    end
  end

  defp parse_raw_expr_mixed_list_error(raw_expr_mixed_list), do:
    raise "#{inspect __MODULE__}.parse_raw_expr_mixed_list parse error. We currently only support the Ecto fragments of the form, 'where: fragment(\"? between ? and ?\", FIELD_AS_VARIABLE, VALUE_AS_VARIABLE, VALUE_AS_VARIABLE)'; and 'where: fragment(\"begins_with(?, ?)\", FIELD_AS_VARIABLE, VALUE_AS_VARIABLE)'. Received: #{inspect raw_expr_mixed_list}"

  defp get_op_clause(left, right, params) do
    field = left |> get_field |> Atom.to_string
    value = get_value(right, params)
    {field, value}
  end

  defp get_field({{:., _, [{:&, _, [0]}, field]}, _, []}), do: field
  defp get_field(other_clause) do
    error "Unsupported where clause, left hand side: #{other_clause}"
  end

  defp get_value({:^, _, [idx]}, params), do: Enum.at(params, idx)
  # HANDLE .all(query) QUERIES
  defp get_value(other_clause, _params), do: other_clause

  defp error(msg) do
    raise ArgumentError, message: msg
  end


  defp extract_select_fields(%Ecto.Query.SelectExpr{expr: expr} = _) do
    case expr do
      {_, _, [0]} ->
        []

      {{:., _, [{_, _, _}, field]}, _, _} ->
        [field]

      {:{}, _, clauses} ->
        for {{_, _, [{_, _, _}, field]}, _, _} <- clauses, do: field
    end
  end

  # Decodes maps and datetime, seemingly unhandled by ExAws Dynamo decoder
  # (timestamps() corresponds with :naive_datetime)
  defp custom_decode(item, model, select) do
    selected_fields = extract_select_fields(select)

    case selected_fields do
      [] ->
        [Enum.reduce(model.__schema__(:fields), item, fn (field, acc) ->
          Map.update!(acc, field, fn val -> decode_type(model.__schema__(:type, field), val) end)
        end)]
      fields ->
        for field <- fields, do: decode_type(model.__schema__(:type, field), Map.get(item, field))
    end
  end

  # This is used slightly differently 
  # when handling select in custom_decode/2
  defp decode_type(type, val) do
    if is_nil val do
      val
    else
      case type do
        :utc_datetime ->
          {:ok, dt, _offset} = DateTime.from_iso8601(val)
          dt

        :naive_datetime ->
          NaiveDateTime.from_iso8601!(val)

        {:embed, _} ->
          decode_embed(type, val)

        t when t in [Ecto.Adapters.DynamoDB.DynamoDBSet, MapSet] ->
          MapSet.new(val)

        _ -> val
      end
    end
  end

  defp decode_embed(type, val) do
    case Ecto.Adapters.SQL.load_embed(type, val) do
      {:ok, decoded_value} ->
        decoded_value
      :error ->
        ecto_dynamo_log(:info, "failed to decode embedded value: #{inspect val}")
        nil
    end
  end
  # We found one instance where DynamoDB's error message could
  # be more instructive - when trying to set an indexed field to something
  # other than a string or number - so we're adding a more helpful message.
  # The parameter, 'params', has the type %{table: :string, records: [:map]}
  defp handle_error!(ex_aws_request_result, params) do
    case ex_aws_request_result do
      {:ok, result}   -> result
      {:error, {error_name, _} = error} ->
        # Check for inappropriate insert into indexed field
        indexed_fields = Ecto.Adapters.DynamoDB.Info.indexed_attributes(params.table)

        # Repo.insert_all can present multiple records at once
        forbidden_insert_on_indexed_field = Enum.reduce(params.records, false, fn (record, acc) -> 
           acc || Enum.any?(record, fn {field, val} ->
            [type] = ExAws.Dynamo.Encoder.encode(val) |> Map.keys
            # Ecto does not convert Empty strings to nil before passing them to Repo.update_all or
            # Repo.insert_all DynamoDB provides an instructive message during an update (forwarded by ExAws),
            # but less so for batch_write_item, so we catch the empty string as well.
            # Dynamo does not allow insertion of empty strings in any case.
            (Enum.member?(indexed_fields, to_string(field)) and not type in ["S", "N"]) || val == ""
          end)
        end)

        cond do 
          # we use this error to check if an update or delete record does not exist
          error_name == "ConditionalCheckFailedException" ->
            {:error, error_name}

          forbidden_insert_on_indexed_field -> 
            raise "The following request error could be related to attempting to insert an empty string or attempting to insert a type other than a string or number on an indexed field. Indexed fields: #{inspect indexed_fields}. Records: #{inspect params.records}.\n\nExAws Request Error! #{inspect error}" 

          true ->
            raise ExAws.Error, message: "ExAws Request Error! #{inspect error}"
        end
    end    
  end

  @doc """
  Logs message to console and optionally to file. Log levels, colours and file path may be set in configuration (details in README.md).
  """
  def ecto_dynamo_log(level, message) do
    colors = Application.get_env(:ecto_adapters_dynamodb, :log_colors)
    d = DateTime.utc_now 
    formatted_message = "\n[Ecto Dynamo #{d.year}-#{d.month}-#{d.day} #{d.hour}:#{d.minute}:#{d.second} UTC] #{message}"
    log_path = Application.get_env(:ecto_adapters_dynamodb, :log_path)
    log_levels = Application.get_env(:ecto_adapters_dynamodb, :log_levels) || [:info]

    if level in log_levels do
      IO.ANSI.format([colors[level] || :normal, formatted_message], true) |> IO.puts

      if String.valid?(log_path) and Regex.match?(~r/\S/, log_path), do: log_pipe(log_path, formatted_message)
    end
  end

  defp log_pipe(path, str) do
    {:ok, file} = File.open(path, [:append])
    IO.binwrite(file, str)
    File.close(file)
  end

end