ecto_adapters_dynamodb

defmodule Ecto.Adapters.DynamoDB do
  @moduledoc """
  Ecto adapter for Amazon DynamoDB.
  """
  @behaviour Ecto.Adapter
  @behaviour Ecto.Adapter.Schema
  @behaviour Ecto.Adapter.Queryable
  @behaviour Ecto.Adapter.Migration

  @impl Ecto.Adapter
  defmacro __before_compile__(_env) do
    # Nothing to see here, yet...
  end

  import Bitwise

  alias Confex.Resolver
  alias Ecto.Adapters.DynamoDB.Cache
  alias Ecto.Adapters.DynamoDB.DynamoDBSet
  alias Ecto.Adapters.DynamoDB.RepoConfig
  alias Ecto.Query.BooleanExpr
  alias ExAws.Dynamo

  require Logger

  @pool_opts [:timeout, :pool_size, :migration_lock]

  # DynamoDB will reject attempts to batch write more than 25 records at once
  # https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html
  @batch_write_item_limit 25

  @impl Ecto.Adapter
  def init(config) do
    log = Keyword.get(config, :log, :debug)
    telemetry_prefix = Keyword.fetch!(config, :telemetry_prefix)

    meta = %{
      opts: Keyword.take(config, @pool_opts),
      telemetry: {config[:repo], log, telemetry_prefix},
      migration_source: Keyword.get(config, :migration_source, "schema_migrations")
    }

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.init", %{
      "#{inspect(__MODULE__)}.init-params" => %{config: config}
    })

    {:ok, Cache.child_spec([config[:repo]]), meta}
  end

  @doc """
  Ensure all applications necessary to run the adapter are started.
  """
  @impl Ecto.Adapter
  def ensure_all_started(config, type) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.ensure_all_started", %{
      "#{inspect(__MODULE__)}.ensure_all_started-params" => %{type: type, config: config}
    })

    with {:ok, _} = Application.ensure_all_started(:ecto_adapters_dynamodb) do
      {:ok, [config]}
    end
  end

  @impl Ecto.Adapter.Migration
  def supports_ddl_transaction?, do: false

  @impl Ecto.Adapter.Migration
  def execute_ddl(adapter_meta, command, options) do
    Ecto.Adapters.DynamoDB.Migration.execute_ddl(adapter_meta, command, options)
  end

  @impl Ecto.Adapter.Migration
  def lock_for_migrations(%{opts: adapter_opts} = _meta, _opts, callback) do
    # TODO - consider adding support for this? See https://github.com/circles-learning-labs/ecto_adapters_dynamodb/issues/34
    if Keyword.get(adapter_opts, :migration_lock) do
      raise "#{inspect(__MODULE__)}.lock_for_migrations error: #{inspect(__MODULE__)} does not currently support migration table lock; please remove the :migration_lock option from your repo configuration or set it to nil"
    else
      callback.()
    end
  end

  @impl Ecto.Adapter
  def checkout(_meta, _opts, _fun) do
    # TODO - consider adding support for this? See https://github.com/circles-learning-labs/ecto_adapters_dynamodb/issues/33
    raise "#{inspect(__MODULE__)}.checkout: #{inspect(__MODULE__)} does not currently support checkout"
  end

  @impl Ecto.Adapter
  def checked_out?(_meta), do: false

  @impl Ecto.Adapter.Queryable
  def stream(_adapter_meta, _query_meta, _query, _params, _opts) do
    # TODO - consider adding support for this? See https://github.com/circles-learning-labs/ecto_adapters_dynamodb/issues/32
    raise "#{inspect(__MODULE__)}.stream: #{inspect(__MODULE__)} does not currently support stream"
  end

  @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()`
  """

  # biggest possible int in 128 bits
  @max_id (1 <<< 128) - 1
  @impl Ecto.Adapter.Schema
  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.

  The adapter builds on ExAws' decoding functionality (see `ExAws.Dynamo.Decoder`) to convert DynamoDB wire format to basic Elixir types. However, Ecto 3.13+ requires additional type conversion from these basic types to proper Ecto types. This module provides custom loaders for :utc_datetime, :naive_datetime, and :decimal types to handle this conversion (e.g., converting string "104.50" to Decimal struct, ISO8601 strings to DateTime structs).
  """
  @impl Ecto.Adapter
  def loaders(:decimal, _type), do: [&load_decimal/1]

  def loaders(type, _type) when type in [:utc_datetime, :utc_datetime_usec],
    do: [&load_utc_datetime/1]

  def loaders(type, _type) when type in [:naive_datetime, :naive_datetime_usec],
    do: [&load_naive_datetime/1]

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

  defp load_decimal(value) when is_binary(value) do
    case Decimal.parse(value) do
      {%Decimal{} = decimal, _rest} -> {:ok, decimal}
      _error -> {:ok, value}
    end
  end

  defp load_decimal(value), do: {:ok, value}

  defp load_utc_datetime(value) when is_binary(value) do
    case DateTime.from_iso8601(value) do
      {:ok, datetime, _offset} -> {:ok, datetime}
      _error -> {:ok, value}
    end
  end

  defp load_utc_datetime(value), do: {:ok, value}

  defp load_naive_datetime(value) when is_binary(value) do
    case NaiveDateTime.from_iso8601(value) do
      {:ok, datetime} -> {:ok, datetime}
      _error -> {:ok, value}
    end
  end

  defp load_naive_datetime(value), do: {:ok, value}

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

  The adapter relies on ExAws encoding functionality to format types for DynamoDB wire protocol. However, some Ecto types need preprocessing before ExAws can handle them. This module provides custom dumpers for :utc_datetime, :naive_datetime (to add ISO8601 formatting), and :decimal types (to convert Decimal structs to strings that ExAws can encode).
  """
  @impl Ecto.Adapter
  def dumpers(type, datetime)
      when type in [:naive_datetime, :naive_datetime_usec, :utc_datetime, :utc_datetime_usec],
      do: [datetime, &to_iso_string/1]

  def dumpers(:decimal, _type), do: [&dump_decimal/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
    iso_string =
      case datetime do
        %NaiveDateTime{} ->
          (datetime |> NaiveDateTime.to_iso8601()) <> "Z"

        %DateTime{} ->
          datetime |> DateTime.to_iso8601()

        nil ->
          nil
      end

    {:ok, iso_string}
  end

  defp dump_decimal(%Decimal{} = decimal), do: {:ok, Decimal.to_string(decimal)}
  defp dump_decimal(value), do: {:ok, value}

  @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}
  @impl Ecto.Adapter.Queryable
  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, "#{inspect(__MODULE__)}.prepare: :all", %{
      "#{inspect(__MODULE__)}.prepare-params" => %{query: inspect(query, structs: false)}
    })

    {:nocache, {:all, query}}
  end

  def prepare(:update_all, query) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.prepare: :update_all", %{
      "#{inspect(__MODULE__)}.prepare-params" => %{query: inspect(query, structs: false)}
    })

    {:nocache, {:update_all, query}}
  end

  def prepare(:delete_all, query) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.prepare: :delete_all", %{
      "#{inspect(__MODULE__)}.prepare-params" => %{query: inspect(query, structs: false)}
    })

    {:nocache, {:delete_all, query}}
  end

  @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}
  @impl Ecto.Adapter.Queryable
  def execute(
        %{repo: repo, migration_source: migration_source},
        query_meta,
        {:nocache, {func, prepared}},
        params,
        opts
      ) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.execute", %{
      "#{inspect(__MODULE__)}.execute-params" => %{
        repo: repo,
        query_meta: query_meta,
        prepared: prepared,
        params: params,
        opts: opts
      }
    })

    # table and model are now nested under .from.source
    {table, model} = prepared.from.source
    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: []

    updated_opts =
      if table == migration_source do
        ecto_dynamo_log(
          :debug,
          "#{inspect(__MODULE__)}.execute: table name corresponds with migration source: #{inspect(migration_source)}. Setting options for recursive scan.",
          %{}
        )

        Keyword.drop(opts, [:timeout, :log, :telemetry_options]) ++ [recursive: true]
      else
        Keyword.drop(opts, [:scan_limit, :limit, :telemetry_options]) ++ scan_limit
      end

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.execute: local variables", %{
      "#{inspect(__MODULE__)}.execute-vars" => %{
        table: table,
        lookup_fields: lookup_fields,
        scan_limit: scan_limit
      }
    })

    case func do
      :delete_all ->
        delete_all(repo, table, lookup_fields, updated_opts)

      :update_all ->
        update_all(repo, table, lookup_fields, updated_opts, prepared.updates, params)

      :all ->
        ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.execute: :all", %{
          "#{inspect(__MODULE__)}.execute-all-vars" => %{
            table: table,
            lookup_fields: lookup_fields,
            updated_opts: updated_opts
          }
        })

        result = Ecto.Adapters.DynamoDB.Query.get_item(repo, table, lookup_fields, updated_opts)

        ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.execute: all: result", %{
          "#{inspect(__MODULE__)}.execute-all-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
          case query_meta do
            %{select: %{from: {_, {_, _, _, types}}}} ->
              types = types_to_source_fields(model, types)
              handle_type_decode(table, result, types, repo, opts)

            _ ->
              if table == migration_source do
                decoded = Enum.map(result["Items"], &decode_item(&1, repo, opts))

                {length(decoded), decoded}
              else
                # Queries with a :select clause will not have the types available in the query_meta,
                # instead construct them from prepared.select
                types = construct_types_from_select_fields(prepared.select)

                handle_type_decode(table, result, types, repo, opts)
              end
          end
        end
    end
  end

  def max_transaction_conflict_retries do
    Application.get_env(
      :ecto_adapters_dynamodb,
      :max_transaction_conflict_retries,
      10
    )
  end

  defp handle_type_decode(table, result, types, repo, opts) do
    if !result["Count"] and !result["Responses"] do
      decoded = decode_item(result["Item"], types, repo, opts)

      {1, [decoded]}
    else
      # 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, &decode_item(&1, types, repo, opts))

      {length(decoded), decoded}
    end
  end

  defp types_to_source_fields(model, types) do
    types
    |> Enum.into([], fn {field, type} ->
      {model.__schema__(:field_source, field), type}
    end)
  end

  # delete_all allows for the recursive option, scanning through multiple pages
  defp delete_all(repo, table, lookup_fields, opts) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.delete_all", %{
      "#{inspect(__MODULE__)}.delete_all-params" => %{
        table: table,
        lookup_fields: lookup_fields,
        opts: opts
      }
    })

    # select only the key
    {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(repo, table)
    scan_or_query = Ecto.Adapters.DynamoDB.Query.scan_or_query?(repo, 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(repo, table, lookup_fields, updated_opts, recursive, %{}, 0)
  end

  defp delete_all_recursive(
         repo,
         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(repo, table, lookup_fields, opts)

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.delete_all_recursive: fetch_result", %{
      "#{inspect(__MODULE__)}.delete_all_recursive-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(repo, 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.RecursiveFetch.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(
        repo,
        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
  # Similarly to a batch insert, batch delete is also restricted by DDB's batch write limit of 25 records - these requests will be chunked as well.
  defp batch_delete(repo, table, prepared_data) do
    Enum.chunk_every(prepared_data, @batch_write_item_limit)
    |> Enum.reduce(%{}, fn batch, unprocessed_items ->
      batch_write_attempt =
        Dynamo.batch_write_item(%{table => batch})
        |> ExAws.request(ex_aws_config(repo))
        |> handle_error!(repo, %{table: table, records: []})

      case batch_write_attempt do
        %{"UnprocessedItems" => %{^table => items}} ->
          Map.update(unprocessed_items, table, items, &(&1 ++ items))

        _ ->
          unprocessed_items
      end
    end)
  end

  defp update_all(repo, table, lookup_fields, opts, updates, params) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.update_all", %{
      "#{inspect(__MODULE__)}.update_all-params" => %{
        table: table,
        lookup_fields: lookup_fields,
        opts: opts
      }
    })

    scan_or_query = Ecto.Adapters.DynamoDB.Query.scan_or_query?(repo, table, lookup_fields)
    recursive = Ecto.Adapters.DynamoDB.Query.parse_recursive_option(scan_or_query, opts)

    key_list = Ecto.Adapters.DynamoDB.Info.primary_key!(repo, table)

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.update_all: key_list", %{
      "#{inspect(__MODULE__)}.update_all-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(repo, updates, params, opts)

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.update_all: update fields", %{
      "#{inspect(__MODULE__)}.update_all-update_fields" => %{
        update_fields_sans_set_remove: inspect(update_fields_sans_set_remove),
        set_remove_fields: inspect(set_remove_fields)
      }
    })

    attribute_names = construct_expression_attribute_names(update_fields_sans_set_remove)

    attribute_values =
      construct_expression_attribute_values(repo, 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(
           repo,
           update_options,
           set_remove_fields,
           opts_with_pull_indexes
         ), []}
      end

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

  defp update_all_recursive(
         repo,
         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(repo, table, lookup_fields, opts)

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.update_all_recursive: fetch_result", %{
      "#{inspect(__MODULE__)}.update_all_recursive-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(repo, table, items, key_list, update_options, set_remove_fields, opts)
      else
        0
      end

    updated_recursive = Ecto.Adapters.DynamoDB.RecursiveFetch.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(
        repo,
        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(repo, 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(
              repo,
              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(ex_aws_config(repo))
        |> handle_error!(repo, %{table: table, records: record ++ []})

        acc + 1
      else
        acc
      end
    end)
  end

  defp update_batch_update_options(repo, 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(repo, set_and_push_fields, opts_with_push)

    set_statement = construct_set_statement(repo, 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(repo, 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
  @impl Ecto.Adapter.Schema
  def insert(repo_meta, schema_meta, fields, on_conflict, returning, opts) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.insert", %{
      "#{inspect(__MODULE__)}.insert-params" => %{
        repo_meta: repo_meta,
        schema_meta: schema_meta,
        fields: fields,
        on_conflict: on_conflict,
        returning: returning,
        opts: opts
      }
    })

    do_insert(repo_meta, schema_meta, fields, on_conflict, opts, 0)
  end

  defp do_insert(repo_meta, schema_meta, fields, on_conflict, opts, retries) do
    if retries >= max_transaction_conflict_retries() do
      raise(
        "#{inspect(__MODULE__)}.insert error: reached maximum transaction conflict retries without success"
      )
    end

    table = schema_meta.source

    model = schema_meta.schema
    fields_map = Enum.into(fields, %{})

    record = maybe_replace_empty_mapsets_for_insert(fields_map, repo_meta.repo, opts)

    insert_nil_fields = opt_config(:insert_nil_fields, repo_meta.repo, opts, true)
    record = unless insert_nil_fields, do: record, else: build_record_map(model, record)

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.insert: local variables", %{
      "#{inspect(__MODULE__)}.insert-vars" => %{table: table, record: record}
    })

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

    replace_all =
      case on_conflict do
        {l, _, _} when is_list(l) ->
          # All fields being set are to be replaced
          Enum.all?(Keyword.keys(fields), &(&1 in l))

        _ ->
          false
      end

    options =
      if replace_all do
        # All fields being replaced: don't use a condition
        []
      else
        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(ex_aws_config(repo_meta.repo))
         |> handle_error!(repo_meta.repo, %{table: table, records: [record]}) do
      {:error, "ConditionalCheckFailedException"} ->
        case on_conflict do
          # Per discussion with Jose Valim (https://github.com/elixir-ecto/ecto/issues/2378)
          # clarifying the adapter should return nothing if there is no `:returning` specified,
          # and what we thought was to be returned as a `nil` id, is only for cases where
          # "the field is autogenerated by the database" (https://hexdocs.pm/ecto/Ecto.Repo.html)
          {:nothing, _, _} ->
            {:ok, []}

          {: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

      {:error, "TransactionConflictException"} ->
        do_insert(repo_meta, schema_meta, fields, on_conflict, opts, retries + 1)

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

  @impl Ecto.Adapter.Schema
  def insert_all(
        %{repo: repo},
        schema_meta,
        field_list,
        rows,
        on_conflict,
        return_sources,
        _placeholders,
        opts
      ) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.insert_all", %{
      "#{inspect(__MODULE__)}.insert_all-params" => %{
        repo: repo,
        schema_meta: schema_meta,
        field_list: field_list,
        rows: rows,
        on_conflict: on_conflict,
        return_sources: return_sources,
        opts: opts
      }
    })

    insert_nil_field_option = Keyword.get(opts, :insert_nil_fields, true)

    do_not_insert_nil_fields =
      insert_nil_field_option == false ||
        RepoConfig.config_val(repo, :insert_nil_fields) == false

    table = schema_meta.source
    model = schema_meta.schema

    prepared_rows =
      Enum.map(rows, fn row ->
        mapped_fields =
          row
          |> Enum.into(%{})
          |> maybe_replace_empty_mapsets_for_insert(repo, opts)

        record =
          if do_not_insert_nil_fields,
            do: mapped_fields,
            else: build_record_map(model, mapped_fields)

        [put_request: [item: record]]
      end)

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.insert_all: local variables", %{
      "#{inspect(__MODULE__)}.insert_all-vars" => %{
        table: table,
        records: get_records_from_fields(prepared_rows)
      }
    })

    batch_write(repo, table, prepared_rows, opts)
  end

  # DynamoDB will reject an entire batch of insert_all() records if there are more than 25 requests.
  # https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html
  # batch_write/4 will break the list into chunks of 25 items and insert each separately.
  defp batch_write(repo, table, prepared_fields, opts) do
    unprocessed_items_element = "UnprocessedItems"
    grouped_records = Enum.chunk_every(prepared_fields, @batch_write_item_limit)
    num_batches = length(grouped_records)

    # Break the prepared_fields into chunks of at most 25 elements to be batch inserted, accumulating
    # the total count of records and appropriate results as it loops through the reduce.
    {total_processed, results} =
      grouped_records
      |> Stream.with_index()
      |> Enum.reduce({0, []}, fn {field_group, i},
                                 {running_total_processed, batch_write_results} ->
        {total_batch_processed, batch_write_attempt} =
          handle_batch_write(repo, field_group, table, unprocessed_items_element)

        # Log depth of 11 will capture the full data structure returned in any UnprocessedItems - https://docs.aws.amazon.com/amazondynamodb/latest/APIReference/API_BatchWriteItem.html
        ecto_dynamo_log(
          :debug,
          "#{inspect(__MODULE__)}.batch_write #{i + 1} of #{num_batches}: local variables",
          %{
            "#{inspect(__MODULE__)}.insert_all-batch_write" => %{
              table: table,
              field_group: field_group,
              results: batch_write_attempt
            }
          },
          depth: 11
        )

        # 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
          query_info = extract_query_info(batch_write_attempt)

          Ecto.Adapters.DynamoDB.QueryInfo.update(opts[:query_info_key], [query_info], fn list ->
            list ++ [query_info]
          end)
        end

        {running_total_processed + total_batch_processed,
         batch_write_results ++ [batch_write_attempt]}
      end)

    result_body_for_log = %{
      table => Enum.flat_map(results, fn res -> res[unprocessed_items_element][table] || [] end)
    }

    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.batch_write: batch_write_attempt result", %{
      "#{inspect(__MODULE__)}.insert_all-batch_write" =>
        inspect(%{
          unprocessed_items_element =>
            if(result_body_for_log[table] == [], do: %{}, else: result_body_for_log)
        })
    })

    {total_processed, nil}
  end

  defp handle_batch_write(repo, field_group, table, unprocessed_items_element) do
    results =
      Dynamo.batch_write_item(%{table => field_group})
      |> ExAws.request(ex_aws_config(repo))
      |> handle_error!(repo, %{table: table, records: get_records_from_fields(field_group)})

    if results[unprocessed_items_element] == %{} do
      {length(field_group), results}
    else
      {length(field_group) - length(results[unprocessed_items_element][table]), results}
    end
  end

  defp get_records_from_fields(fields),
    do: Enum.map(fields, fn [put_request: [item: record]] -> record 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.
    fields = model.__schema__(:fields)
    sources = fields |> Enum.into(%{}, fn f -> {f, model.__schema__(:field_source, f)} end)

    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()
    |> Enum.reduce(%{}, fn {k, v}, acc ->
      Map.put(acc, Map.get(sources, k), v)
    end)
    |> Map.merge(empty_strings_to_nil)
  end

  @impl Ecto.Adapter.Schema
  def delete(adapter_meta = %{repo: repo}, schema_meta, filters, _returning, opts) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.delete", %{
      "#{inspect(__MODULE__)}.delete-params" => %{
        adapter_meta: adapter_meta,
        schema_meta: schema_meta,
        filters: filters,
        opts: opts
      }
    })

    do_delete(repo, schema_meta, filters, opts, 0)
  end

  defp do_delete(repo, schema_meta, filters, opts, retries) do
    if retries >= max_transaction_conflict_retries() do
      raise(
        "#{inspect(__MODULE__)}.delete error: reached maximum transaction conflict retries without success"
      )
    end

    table = schema_meta.source

    updated_filters =
      maybe_update_filters_for_range_key(repo, table, schema_meta, filters, opts, "delete")

    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(ex_aws_config(repo))
         |> handle_error!(repo, %{table: table, records: record ++ []}) do
      %{} ->
        {:ok, []}

      {:error, "ConditionalCheckFailedException"} ->
        {:error, :stale}

      {:error, "TransactionConflictException"} ->
        do_delete(repo, schema_meta, filters, opts, retries + 1)
    end
  end

  @impl Ecto.Adapter.Schema
  def update(adapter_meta = %{repo: repo}, schema_meta, fields, filters, returning, opts) do
    ecto_dynamo_log(:debug, "#{inspect(__MODULE__)}.update", %{
      "#{inspect(__MODULE__)}.update-params" => %{
        adapter_meta: adapter_meta,
        schema_meta: schema_meta,
        fields: fields,
        filters: filters,
        returning: returning,
        opts: opts
      }
    })

    do_update(repo, schema_meta, fields, filters, returning, opts, 0)
  end

  defp do_update(_repo, _schema_meta, _fields, _filters, _returning, _opts, retries)

  defp do_update(repo, schema_meta, fields, filters, returning, opts, retries) do
    if retries >= max_transaction_conflict_retries() do
      raise(
        "#{inspect(__MODULE__)}.update error: reached maximum transaction conflict retries without success"
      )
    end

    table = schema_meta.source

    updated_filters =
      maybe_update_filters_for_range_key(repo, table, schema_meta, filters, opts, "update")

    fields = maybe_replace_empty_mapsets_for_update(fields, repo, opts)

    update_expression = construct_update_expression(repo, 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(repo, 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(ex_aws_config(repo))
         |> handle_error!(repo, %{table: table, records: record ++ []}) do
      %{} ->
        {:ok, []}

      {:error, "ConditionalCheckFailedException"} ->
        {:error, :stale}

      {:error, "TransactionConflictException"} ->
        do_update(repo, schema_meta, fields, filters, returning, opts, retries + 1)
    end
  end

  # Support for tables with a hash+range key.
  #
  #  * If the schema has both keys declared (using the `primary_key: true`) the filters are already correct
  #  * If :range_key is specified with a value it is added to filters
  #  * If :range_key is not specified, and the table does have a range key, attempt to find it with a DynamoDB query
  #
  defp maybe_update_filters_for_range_key(repo, table, schema_meta, filters, opts, action) do
    with primary_key_length <- length(schema_meta.schema.__schema__(:primary_key)) do
      case opts[:range_key] do
        # Use primary keys declared in schema
        nil when primary_key_length == 2 ->
          filters

        nil ->
          {:primary, key_list} = Ecto.Adapters.DynamoDB.Info.primary_key!(repo, 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(repo, table, filters_as_strings, updated_opts)

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

            if items == [],
              do:
                raise(
                  "#{inspect(__MODULE__)}.#{action} error: no results found for record: #{inspect(filters)}"
                )

            if length(items) > 1,
              do:
                raise(
                  "#{inspect(__MODULE__)}.#{action} error: more than one result found for record: #{inspect(filters)} Please consider using the adapter's :range_key custom inline option (see README)."
                )

            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
    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(filters) when is_list(filters) do
    Keyword.keys(filters)
    |> Enum.map(fn field -> "attribute_exists(##{to_string(field)})" end)
    |> Enum.join(" AND ")
  end

  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(repo, fields, opts) do
    remove_rather_than_set_to_null =
      opts[:remove_nil_fields] || opts[:remove_nil_fields_on_update] ||
        RepoConfig.config_val(repo, :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:

    fields
    |> Enum.reduce([], &format_update_field(&1, &2, remove_rather_than_set_to_null, opts))
    |> Enum.filter(fn {x, _} -> not Keyword.has_key?(maybe_list(opts[:pull]), x) end)
  end

  defp format_update_field({_k, nil}, acc, true, _opts), do: acc
  defp format_update_field({k, v}, acc, true, opts), do: [{k, format_val(k, v, opts)} | acc]

  defp format_update_field({k, v}, acc, false, opts),
    do: [{k, format_nil_or_val(k, v, opts)} | acc]

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

  defp format_nil_or_val(_k, nil, _opts), do: %{"NULL" => "true"}
  defp format_nil_or_val(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(_repo, 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 parameters
  # whereas update_all includes a more complicated updates structure
  defp construct_update_expression(repo, fields, opts) do
    set_statement = construct_set_statement(repo, fields, opts)
    rem_statement = construct_remove_statement(repo, fields, opts)

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

  # fields::[{:field, val}]
  defp construct_set_statement(repo, fields, opts) do
    remove_rather_than_set_to_null =
      opts[:remove_nil_fields] || opts[:remove_nil_fields_on_update] ||
        RepoConfig.config_val(repo, :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(repo, fields, opts) do
    remove_rather_than_set_to_null =
      opts[:remove_nil_fields] || opts[:remove_nil_fields_on_update] ||
        RepoConfig.config_val(repo, :remove_nil_fields_on_update) == true

    # Ecto :pull update can be emulated provided
    # we are given an index to remove in opts[:pull_indexes]
    remove_clauses =
      if remove_rather_than_set_to_null do
        for {key, val} <- fields, is_nil(val), do: "##{Atom.to_string(key)}"
      else
        []
      end ++
        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 queries with variable values, ex. Repo.all from i in Item, where: i.id in ^item_ids
  # The last element of the tuple (first arg) will be a list with two numbers;
  # the first number will be the number of attributes to be updated (in the event of an update_all query with a variable list)
  # and the second will be a count of the number of elements in the variable list being queried. For example:
  #
  # query = from p in Person, where: p.id in ^ids
  # TestRepo.update_all(query, set: [password: "cheese", last_name: "Smith"])
  #
  # assuming that ids contains 4 values, the last element would be [2, 4].
  # Use this data to modify the params, which would otherwise include the values to be updated as well, which we don't want to query on.
  defp get_value({:^, _, [num_update_terms, _num_query_terms]}, params),
    do: Enum.drop(params, num_update_terms)

  # Seems to be necessary for handling running a batch of migrations down.
  defp get_value(%{value: right}, params), do: get_value(right, params)
  # Handle .all(query) queries
  defp get_value(other_clause, _params), do: other_clause

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

  defp construct_types_from_select_fields(%Ecto.Query.SelectExpr{expr: expr}) do
    case expr do
      {:{}, [], clauses = [{{:., [type: _type], [{:&, [], [0]}, _field]}, [], []} | _]} ->
        for {{:., [type: type], [{:&, [], [0]}, field]}, [], []} <- clauses, do: {field, type}

      {_, _, [0]} ->
        []

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

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

  def decode_item(item, types, repo, opts) do
    types
    |> Enum.map(fn {field, type} ->
      Map.get(item, Atom.to_string(field), %{"NULL" => true})
      |> Dynamo.Decoder.decode()
      |> decode_type(type, repo, opts)
    end)
  end

  def decode_item(%{"version" => version}, _repo, _opts) do
    [version |> Dynamo.Decoder.decode()]
  end

  # Decodes datetime, seemingly unhandled by ExAws Dynamo decoder
  defp decode_type(nil, DynamoDBSet, repo, opts), do: maybe_replace_nil_mapset(repo, opts)

  defp decode_type(nil, _type, _repo, _opts), do: nil

  defp decode_type(val, type, _repo, _opts) when type in [:utc_datetime_usec, :utc_datetime] do
    {:ok, dt, _offset} = DateTime.from_iso8601(val)
    dt
  end

  defp decode_type(val, type, _repo, _opts) when type in [:naive_datetime_usec, :naive_datetime],
    do: NaiveDateTime.from_iso8601!(val)

  # Support for Ecto >= 3.5
  defp decode_type(val, {:parameterized, Ecto.Embedded, _} = type, _repo, _opts),
    do: decode_embed(val, type)

  # Support for Ecto 3.0-3.4
  defp decode_type(val, {:embed, _} = type, _repo, _opts), do: decode_embed(val, type)

  # Support for Ecto >= 3.13
  defp decode_type(val, :decimal, _repo, _opts) when is_binary(val) do
    case Decimal.parse(val) do
      {%Decimal{} = decimal, _rest} -> decimal
      _error -> val
    end
  end

  defp decode_type(val, _type, _repo, _opts), do: val

  defp decode_embed(val, type) do
    try do
      case Ecto.Type.embedded_load(type, val, :json) do
        {:ok, decoded_value} ->
          handle_decoded_embedded(decoded_value)

        :error ->
          ecto_dynamo_log(
            :debug,
            "#{inspect(__MODULE__)}.decode_embed: failed to decode embedded value: #{inspect(val)}"
          )

          nil
      end
    rescue
      _ ->
        ecto_dynamo_log(
          :debug,
          "#{inspect(__MODULE__)}.decode_embed: failed to decode embedded value: #{inspect(val)}"
        )

        nil
    end
  end

  defp handle_decoded_embedded(embedded) when is_list(embedded),
    do: Enum.map(embedded, &unload_parameterized_fields/1)

  defp handle_decoded_embedded(embedded), do: unload_parameterized_fields(embedded)

  # Rebuilds the embedded schema unloading the parameterized fields, so the parameterized
  # type can load it in the ecto schema.
  defp unload_parameterized_fields(%schema{} = embedded) do
    fields = schema.__schema__(:fields)

    Enum.reduce(fields, embedded, fn field, acc ->
      field_type = schema.__schema__(:type, field)
      field_value = Map.get(embedded, field)

      Map.put(acc, field, maybe_dump_field(field_value, field_type))
    end)
  end

  defp maybe_dump_field(val, {:parameterized, _type, _params} = field_type) do
    {:ok, unloaded_value} = Ecto.Type.dump(field_type, val)
    unloaded_value
  end

  defp maybe_dump_field(val, _field_type), do: val

  # 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!({:ok, result}, _repo, _params), do: result

  defp handle_error!({:error, {error_name, _} = error}, repo, params) do
    # Check for inappropriate insert into indexed field
    indexed_fields = Ecto.Adapters.DynamoDB.Info.indexed_attributes(repo, 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] = 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 type not 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 in ["ConditionalCheckFailedException", "TransactionConflictException"] ->
        {: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

  @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, attributes \\ %{}, opts \\ []) do
    if Confex.get_env(:ecto_adapters_dynamodb, :use_logger) do
      Logger.log(level, message, attributes)
    else
      write_console_log(level, message, attributes, opts)
    end
  end

  defp write_console_log(level, message, attributes, opts) do
    log_levels = Confex.get_env(:ecto_adapters_dynamodb, :log_levels) || [:info]

    if level in log_levels do
      log_path = Confex.get_env(:ecto_adapters_dynamodb, :log_path)
      depth = opts[:depth] || 4
      colours = Confex.get_env(:ecto_adapters_dynamodb, :log_colours)
      d = DateTime.utc_now()

      formatted_message =
        "#{d.year}-#{d.month}-#{d.day} #{d.hour}:#{d.minute}:#{d.second} UTC [Ecto dynamo #{level}] #{inspect(message)}"

      {:ok, log_message} =
        Jason.encode(%{message: formatted_message, attributes: chisel(attributes, depth)})

      if Confex.get_env(:ecto_adapters_dynamodb, :log_in_colour) do
        IO.ANSI.format([colours[level] || :normal, log_message], true) |> IO.puts()
      else
        log_message |> IO.puts()
      end

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

  def ex_aws_config(repo) do
    config = Resolver.resolve!(repo.config())

    config
    |> Keyword.take([:debug_requests, :access_key_id, :secret_access_key, :region])
    |> Keyword.merge(Keyword.get(config, :dynamodb, []))
  end

  defp chisel(str, _depth) when is_binary(str),
    do: if(String.valid?(str), do: str, else: Base.encode64(str))

  defp chisel(num, _depth) when is_number(num), do: num
  defp chisel(any, _depth) when not is_map(any) and not is_list(any), do: inspect(any)
  defp chisel(_, 0), do: "beyond_log_depth"
  defp chisel(%{__struct__: _} = struct, _depth), do: inspect(struct)

  defp chisel(map, depth) when is_map(map) do
    for {k, v} <- map, into: %{}, do: {k, chisel(v, depth - 1)}
  end

  defp chisel(list, depth) when is_list(list) do
    for e <- list, do: chisel(e, depth - 1)

    # Stream.with_index(list) |> Enum.reduce(%{}, fn({v,k}, acc)-> Map.put(acc, k, chisel(v, depth - 1)) end)
  end

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

  defp opt_config(key, repo, opts, default \\ false) do
    case Keyword.get(opts, key) do
      nil -> RepoConfig.config_val(repo, key, default)
      x -> x
    end
  end

  defp maybe_replace_empty_mapsets_for_insert(record, repo, opts) do
    empty_mapset_to_nil = opt_config(:empty_mapset_to_nil, repo, opts)
    insert_nil = opt_config(:insert_nil_fields, repo, opts, true)

    cond do
      empty_mapset_to_nil and insert_nil ->
        record
        |> Enum.map(fn {k, v} -> {k, empty_mapset_to_nil(v)} end)
        |> Enum.into(%{})

      empty_mapset_to_nil ->
        record
        |> Enum.reject(fn {_k, v} -> v == MapSet.new() end)
        |> Enum.into(%{})

      true ->
        record
    end
  end

  defp maybe_replace_empty_mapsets_for_update(record, repo, opts) do
    if opt_config(:empty_mapset_to_nil, repo, opts) do
      record
      |> Enum.map(fn {k, v} -> {k, empty_mapset_to_nil(v)} end)
    else
      record
    end
  end

  defp empty_mapset_to_nil(%MapSet{} = m), do: if(MapSet.size(m) == 0, do: nil, else: m)

  defp empty_mapset_to_nil(x), do: x

  defp maybe_replace_nil_mapset(repo, opts) do
    if opt_config(:nil_to_empty_mapset, repo, opts) do
      MapSet.new()
    else
      nil
    end
  end
end