ex_credstash

# Copyright 2026 Relay, Inc.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

defmodule ExCredstash.Dynamo do
  @moduledoc """
  DynamoDB operations for credential storage.

  This module provides functions for interacting with AWS DynamoDB
  to store and retrieve encrypted credentials using the official
  `:aws` Erlang SDK.

  ## Table Schema

  The credential store table uses:
  - Partition key: `name` (String) - The secret name
  - Sort key: `version` (String) - Zero-padded version number (19 chars)

  ## Item Structure

  Each item contains:
  | Attribute | Type | Description |
  |-----------|------|-------------|
  | `name` | String | Secret name (partition key) |
  | `version` | String | Zero-padded version, e.g., "0000000000000000001" |
  | `key` | String | Base64-encoded KMS-encrypted data key |
  | `contents` | String | Base64-encoded AES-encrypted secret |
  | `hmac` | Binary | Hex-encoded HMAC of ciphertext |
  | `digest` | String | Hash algorithm (e.g., "SHA256") |
  | `comment` | String | Optional comment |

  ## Usage

      # Create a client
      client = ExCredstash.Dynamo.client(region: "us-east-1")

      # Store a secret
      {:ok, :created} = ExCredstash.Dynamo.put_secret(
        region: "us-east-1",
        name: "my_secret",
        version: "0000000000000000001",
        key: "base64_encoded_key",
        contents: "base64_encoded_contents",
        hmac: "hex_encoded_hmac",
        digest: "SHA256"
      )

      # Get the latest version
      {:ok, item} = ExCredstash.Dynamo.get_latest_secret(
        region: "us-east-1",
        name: "my_secret"
      )

  """

  @default_table "credential-store"

  @doc """
  Create an AWS DynamoDB client.

  Uses `:aws_credentials` to get credentials automatically from the
  credential chain (env vars, instance profile, etc.).

  ## Options

    * `:region` - AWS region (required)
    * `:access_key_id` - Override access key (optional)
    * `:secret_access_key` - Override secret key (optional)
    * `:session_token` - Override session token (optional)

  ## Examples

      # Using default credentials
      client = ExCredstash.Dynamo.client(region: "us-east-1")

      # With explicit credentials
      client = ExCredstash.Dynamo.client(
        region: "us-east-1",
        access_key_id: "AKIAIOSFODNN7EXAMPLE",
        secret_access_key: "wJalrXUtnFEMI/K7MDENG/bPxRfiCYEXAMPLEKEY"
      )

  """
  @spec client(keyword()) :: AWS.Client.t() | {:error, term()}
  def client(opts) do
    region = get_region(opts)

    cond do
      is_nil(region) ->
        {:error, {:configuration_error, "AWS region is required"}}

      opts[:access_key_id] && opts[:secret_access_key] ->
        # Use explicitly provided credentials
        AWS.Client.create(
          opts[:access_key_id],
          opts[:secret_access_key],
          opts[:session_token],
          region
        )

      true ->
        # Use aws_credentials to get credentials from the environment
        case get_credentials() do
          {:ok, creds} ->
            AWS.Client.create(
              creds.access_key_id,
              creds.secret_access_key,
              Map.get(creds, :token),
              region
            )

          {:error, _reason} = error ->
            error
        end
    end
  end

  @doc """
  Create the credential-store table if it doesn't exist.

  Uses PAY_PER_REQUEST billing mode by default (on-demand capacity).

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:tags` - Map of tags to apply to the table (optional)

  ## Returns

    * `{:ok, :created}` - Table was created
    * `{:ok, :exists}` - Table already exists
    * `{:error, reason}` - Error occurred

  ## Examples

      # Create with defaults
      {:ok, :created} = ExCredstash.Dynamo.create_table(region: "us-east-1")

      # Create with custom name and tags
      {:ok, _} = ExCredstash.Dynamo.create_table(
        region: "us-east-1",
        table: "my-credentials",
        tags: %{"Environment" => "production"}
      )

  """
  @spec create_table(keyword()) :: {:ok, :created | :exists} | {:error, term()}
  def create_table(opts) do
    case client(opts) do
      {:error, _reason} = error ->
        error

      aws_client ->
        table = opts[:table] || @default_table

        # First check if table exists
        case AWS.DynamoDB.describe_table(aws_client, %{"TableName" => table}) do
          {:ok, _response, _http_response} ->
            {:ok, :exists}

          {:error, {:unexpected_response, %{body: body}}} ->
            if is_resource_not_found?(body) do
              # Table doesn't exist, create it
              do_create_table(aws_client, table, opts[:tags])
            else
              parse_aws_error(body, :describe_table)
            end

          {:error, reason} ->
            {:error, {:dynamo_error, :describe_table, reason}}
        end
    end
  end

  @doc """
  Put a secret item into the table.

  Uses a conditional expression to prevent overwriting existing versions.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:name` - Secret name (required)
    * `:version` - Version string, must be pre-padded (required)
    * `:key` - Base64-encoded encrypted data key (required)
    * `:contents` - Base64-encoded encrypted secret (required)
    * `:hmac` - Hex-encoded HMAC (required)
    * `:digest` - Hash algorithm string (required)
    * `:comment` - Optional comment

  ## Returns

    * `{:ok, :created}` - Item created successfully
    * `{:error, {:already_exists, version}}` - Version already exists
    * `{:error, reason}` - Other error

  ## Examples

      {:ok, :created} = ExCredstash.Dynamo.put_secret(
        region: "us-east-1",
        name: "db_password",
        version: "0000000000000000001",
        key: "base64_key_data",
        contents: "base64_encrypted_content",
        hmac: "hexhmacvalue",
        digest: "SHA256",
        comment: "Database password for production"
      )

  """
  @spec put_secret(keyword()) :: {:ok, :created} | {:error, term()}
  def put_secret(opts) do
    with :ok <- validate_required(opts, [:name, :version, :key, :contents, :hmac, :digest]) do
      case client(opts) do
        {:error, _reason} = error ->
          error

        aws_client ->
          table = opts[:table] || @default_table
          item = build_put_item(opts)

          request = %{
            "TableName" => table,
            "Item" => item,
            "ConditionExpression" => "attribute_not_exists(#name)",
            "ExpressionAttributeNames" => %{"#name" => "name"}
          }

          case AWS.DynamoDB.put_item(aws_client, request) do
            {:ok, _response, _http_response} ->
              {:ok, :created}

            {:error, {:unexpected_response, %{body: body}}} ->
              if is_conditional_check_failed?(body) do
                # Include the version that already exists in the error
                {:error, {:already_exists, opts[:version]}}
              else
                parse_aws_error(body, :put_item)
              end

            {:error, reason} ->
              {:error, {:dynamo_error, :put_item, reason}}
          end
      end
    end
  end

  @doc """
  Get a specific version of a secret.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:name` - Secret name (required)
    * `:version` - Version string (required)

  ## Returns

    * `{:ok, item_map}` - The secret item as a map
    * `{:error, :not_found}` - Item not found
    * `{:error, reason}` - Other error

  ## Examples

      {:ok, item} = ExCredstash.Dynamo.get_secret(
        region: "us-east-1",
        name: "db_password",
        version: "0000000000000000001"
      )

  """
  @spec get_secret(keyword()) :: {:ok, map()} | {:error, term()}
  def get_secret(opts) do
    with :ok <- validate_required(opts, [:name, :version]) do
      case client(opts) do
        {:error, _reason} = error ->
          error

        aws_client ->
          table = opts[:table] || @default_table
          name = opts[:name]
          version = opts[:version]

          request = %{
            "TableName" => table,
            "Key" => %{
              "name" => %{"S" => name},
              "version" => %{"S" => version}
            },
            "ConsistentRead" => true
          }

          case AWS.DynamoDB.get_item(aws_client, request) do
            {:ok, response, _http_response} ->
              case Map.get(response, "Item") do
                nil -> {:error, :not_found}
                item -> {:ok, parse_item(item)}
              end

            {:error, {:unexpected_response, %{body: body}}} ->
              parse_aws_error(body, :get_item)

            {:error, reason} ->
              {:error, {:dynamo_error, :get_item, reason}}
          end
      end
    end
  end

  @doc """
  Get the latest version of a secret.

  Uses Query with ScanIndexForward=false and Limit=1 to efficiently
  retrieve only the highest version.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:name` - Secret name (required)

  ## Returns

    * `{:ok, item_map}` - The latest secret item
    * `{:error, :not_found}` - No versions exist
    * `{:error, reason}` - Other error

  ## Examples

      {:ok, item} = ExCredstash.Dynamo.get_latest_secret(
        region: "us-east-1",
        name: "db_password"
      )

  """
  @spec get_latest_secret(keyword()) :: {:ok, map()} | {:error, term()}
  def get_latest_secret(opts) do
    with :ok <- validate_required(opts, [:name]) do
      case client(opts) do
        {:error, _reason} = error ->
          error

        aws_client ->
          table = opts[:table] || @default_table
          name = opts[:name]

          request = %{
            "TableName" => table,
            "KeyConditionExpression" => "#name = :name",
            "ExpressionAttributeNames" => %{"#name" => "name"},
            "ExpressionAttributeValues" => %{":name" => %{"S" => name}},
            "ScanIndexForward" => false,
            "Limit" => 1,
            "ConsistentRead" => true
          }

          case AWS.DynamoDB.query(aws_client, request) do
            {:ok, response, _http_response} ->
              case Map.get(response, "Items", []) do
                [] -> {:error, :not_found}
                [item | _] -> {:ok, parse_item(item)}
              end

            {:error, {:unexpected_response, %{body: body}}} ->
              parse_aws_error(body, :query)

            {:error, reason} ->
              {:error, {:dynamo_error, :query, reason}}
          end
      end
    end
  end

  @doc """
  Get the highest version number for a secret.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:name` - Secret name (required)

  ## Returns

    * `{:ok, version_integer}` - The highest version as integer
    * `{:ok, 0}` - No versions exist
    * `{:error, reason}` - Error occurred

  ## Examples

      {:ok, 3} = ExCredstash.Dynamo.get_highest_version(
        region: "us-east-1",
        name: "db_password"
      )

      # Secret doesn't exist
      {:ok, 0} = ExCredstash.Dynamo.get_highest_version(
        region: "us-east-1",
        name: "nonexistent"
      )

  """
  @spec get_highest_version(keyword()) :: {:ok, non_neg_integer()} | {:error, term()}
  def get_highest_version(opts) do
    with :ok <- validate_required(opts, [:name]) do
      case client(opts) do
        {:error, _reason} = error ->
          error

        aws_client ->
          table = opts[:table] || @default_table
          name = opts[:name]

          request = %{
            "TableName" => table,
            "KeyConditionExpression" => "#name = :name",
            "ExpressionAttributeNames" => %{"#name" => "name"},
            "ExpressionAttributeValues" => %{":name" => %{"S" => name}},
            "ScanIndexForward" => false,
            "Limit" => 1,
            "ConsistentRead" => true,
            "ProjectionExpression" => "version"
          }

          case AWS.DynamoDB.query(aws_client, request) do
            {:ok, response, _http_response} ->
              case Map.get(response, "Items", []) do
                [] ->
                  {:ok, 0}

                [item | _] ->
                  version_str = get_string(item, "version") || "0"
                  {:ok, String.to_integer(version_str)}
              end

            {:error, {:unexpected_response, %{body: body}}} ->
              parse_aws_error(body, :query)

            {:error, reason} ->
              {:error, {:dynamo_error, :query, reason}}
          end
      end
    end
  end

  @doc """
  List all secrets in the table.

  Returns name, version, and comment for each item. Uses pagination
  to handle tables with many items.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")

  ## Returns

    * `{:ok, [%{name: String.t(), version: String.t(), comment: String.t() | nil}]}`
    * `{:error, reason}`

  ## Examples

      {:ok, secrets} = ExCredstash.Dynamo.list_secrets(region: "us-east-1")
      # [%{name: "db_password", version: "0000000000000000001", comment: nil}, ...]

  """
  @spec list_secrets(keyword()) :: {:ok, list(map())} | {:error, term()}
  def list_secrets(opts) do
    case client(opts) do
      {:error, _reason} = error ->
        error

      aws_client ->
        table = opts[:table] || @default_table
        do_scan_all(aws_client, table, nil, [])
    end
  end

  @doc """
  Get all unique secret names in the table.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")

  ## Returns

    * `{:ok, [String.t()]}` - List of unique names, sorted alphabetically
    * `{:error, reason}`

  ## Examples

      {:ok, names} = ExCredstash.Dynamo.list_names(region: "us-east-1")
      # ["api_key", "db_password", "secret_token"]

  """
  @spec list_names(keyword()) :: {:ok, list(String.t())} | {:error, term()}
  def list_names(opts) do
    case list_secrets(opts) do
      {:ok, secrets} ->
        names =
          secrets
          |> Enum.map(& &1.name)
          |> Enum.uniq()
          |> Enum.sort()

        {:ok, names}

      {:error, _reason} = error ->
        error
    end
  end

  @doc """
  Delete all versions of a secret.

  Queries for all versions and deletes them one by one.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:name` - Secret name (required)

  ## Returns

    * `{:ok, deleted_count}` - Number of items deleted
    * `{:error, reason}`

  ## Examples

      {:ok, 3} = ExCredstash.Dynamo.delete_secret(
        region: "us-east-1",
        name: "old_secret"
      )

  """
  @spec delete_secret(keyword()) :: {:ok, non_neg_integer()} | {:error, term()}
  def delete_secret(opts) do
    with :ok <- validate_required(opts, [:name]) do
      case client(opts) do
        {:error, _reason} = error ->
          error

        aws_client ->
          table = opts[:table] || @default_table
          name = opts[:name]

          # First query all versions
          case do_query_all_versions(aws_client, table, name, nil, []) do
            {:ok, versions} ->
              # Delete each version
              do_delete_all(aws_client, table, versions, 0)

            {:error, _reason} = error ->
              error
          end
      end
    end
  end

  @doc """
  Query all versions of a specific secret.

  ## Options

    * `:region` - AWS region (required)
    * `:table` - Table name (default: "credential-store")
    * `:name` - Secret name (required)

  ## Returns

    * `{:ok, [item_map]}` - List of all versions
    * `{:error, reason}`

  ## Examples

      {:ok, versions} = ExCredstash.Dynamo.query_secret_versions(
        region: "us-east-1",
        name: "db_password"
      )

  """
  @spec query_secret_versions(keyword()) :: {:ok, list(map())} | {:error, term()}
  def query_secret_versions(opts) do
    with :ok <- validate_required(opts, [:name]) do
      case client(opts) do
        {:error, _reason} = error ->
          error

        aws_client ->
          table = opts[:table] || @default_table
          name = opts[:name]

          do_query_all_versions_full(aws_client, table, name, nil, [])
      end
    end
  end

  # Private Functions

  defp get_region(opts) do
    opts[:region] ||
      Application.get_env(:ex_credstash, :region) ||
      System.get_env("AWS_DEFAULT_REGION") ||
      System.get_env("AWS_REGION")
  end

  defp get_credentials do
    try do
      creds = :aws_credentials.get_credentials()

      case creds do
        :undefined ->
          {:error, {:credentials_error, :no_credentials_available}}

        creds when is_map(creds) ->
          {:ok, creds}

        other ->
          {:error, {:credentials_error, {:unexpected_response, other}}}
      end
    catch
      :exit, reason ->
        {:error, {:credentials_error, {:exit, reason}}}

      :error, reason ->
        {:error, {:credentials_error, {:error, reason}}}

      kind, reason ->
        {:error, {:credentials_error, {kind, reason}}}
    end
  end

  defp validate_required(opts, required_keys) do
    missing =
      Enum.filter(required_keys, fn key ->
        is_nil(opts[key])
      end)

    case missing do
      [] -> :ok
      keys -> {:error, {:missing_required, keys}}
    end
  end

  defp do_create_table(aws_client, table, tags) do
    request = %{
      "TableName" => table,
      "AttributeDefinitions" => [
        %{"AttributeName" => "name", "AttributeType" => "S"},
        %{"AttributeName" => "version", "AttributeType" => "S"}
      ],
      "KeySchema" => [
        %{"AttributeName" => "name", "KeyType" => "HASH"},
        %{"AttributeName" => "version", "KeyType" => "RANGE"}
      ],
      "BillingMode" => "PAY_PER_REQUEST"
    }

    # Add tags if provided
    request =
      if tags && map_size(tags) > 0 do
        tag_list =
          Enum.map(tags, fn {key, value} ->
            %{"Key" => key, "Value" => value}
          end)

        # Add default credstash tag
        tag_list = [%{"Key" => "Name", "Value" => "credstash"} | tag_list]

        Map.put(request, "Tags", tag_list)
      else
        Map.put(request, "Tags", [%{"Key" => "Name", "Value" => "credstash"}])
      end

    case AWS.DynamoDB.create_table(aws_client, request) do
      {:ok, _response, _http_response} ->
        {:ok, :created}

      {:error, {:unexpected_response, %{body: body}}} ->
        parse_aws_error(body, :create_table)

      {:error, reason} ->
        {:error, {:dynamo_error, :create_table, reason}}
    end
  end

  defp build_put_item(opts) do
    # HMAC is stored as Binary type (B)
    # The hex string is Base64-encoded for the Binary type.
    hmac_b64 = Base.encode64(opts[:hmac])

    item = %{
      "name" => %{"S" => opts[:name]},
      "version" => %{"S" => opts[:version]},
      "key" => %{"S" => opts[:key]},
      "contents" => %{"S" => opts[:contents]},
      "hmac" => %{"B" => hmac_b64},
      "digest" => %{"S" => opts[:digest]}
    }

    # Add optional comment
    if opts[:comment] && opts[:comment] != "" do
      Map.put(item, "comment", %{"S" => opts[:comment]})
    else
      item
    end
  end

  defp do_scan_all(aws_client, table, exclusive_start_key, acc) do
    request = %{
      "TableName" => table,
      "ProjectionExpression" => "#name, version, #comment",
      "ExpressionAttributeNames" => %{
        "#name" => "name",
        "#comment" => "comment"
      }
    }

    request =
      if exclusive_start_key do
        Map.put(request, "ExclusiveStartKey", exclusive_start_key)
      else
        request
      end

    case AWS.DynamoDB.scan(aws_client, request) do
      {:ok, response, _http_response} ->
        items =
          response
          |> Map.get("Items", [])
          |> Enum.map(&parse_list_item/1)

        new_acc = acc ++ items

        case Map.get(response, "LastEvaluatedKey") do
          nil ->
            {:ok, new_acc}

          last_key ->
            do_scan_all(aws_client, table, last_key, new_acc)
        end

      {:error, {:unexpected_response, %{body: body}}} ->
        parse_aws_error(body, :scan)

      {:error, reason} ->
        {:error, {:dynamo_error, :scan, reason}}
    end
  end

  defp do_query_all_versions(aws_client, table, name, exclusive_start_key, acc) do
    request = %{
      "TableName" => table,
      "KeyConditionExpression" => "#name = :name",
      "ExpressionAttributeNames" => %{"#name" => "name"},
      "ExpressionAttributeValues" => %{":name" => %{"S" => name}},
      "ProjectionExpression" => "#name, version"
    }

    request =
      if exclusive_start_key do
        Map.put(request, "ExclusiveStartKey", exclusive_start_key)
      else
        request
      end

    case AWS.DynamoDB.query(aws_client, request) do
      {:ok, response, _http_response} ->
        items =
          response
          |> Map.get("Items", [])
          |> Enum.map(fn item ->
            %{
              name: get_string(item, "name"),
              version: get_string(item, "version")
            }
          end)

        new_acc = acc ++ items

        case Map.get(response, "LastEvaluatedKey") do
          nil ->
            {:ok, new_acc}

          last_key ->
            do_query_all_versions(aws_client, table, name, last_key, new_acc)
        end

      {:error, {:unexpected_response, %{body: body}}} ->
        parse_aws_error(body, :query)

      {:error, reason} ->
        {:error, {:dynamo_error, :query, reason}}
    end
  end

  defp do_query_all_versions_full(aws_client, table, name, exclusive_start_key, acc) do
    request = %{
      "TableName" => table,
      "KeyConditionExpression" => "#name = :name",
      "ExpressionAttributeNames" => %{"#name" => "name"},
      "ExpressionAttributeValues" => %{":name" => %{"S" => name}},
      "ConsistentRead" => true
    }

    request =
      if exclusive_start_key do
        Map.put(request, "ExclusiveStartKey", exclusive_start_key)
      else
        request
      end

    case AWS.DynamoDB.query(aws_client, request) do
      {:ok, response, _http_response} ->
        items =
          response
          |> Map.get("Items", [])
          |> Enum.map(&parse_item/1)

        new_acc = acc ++ items

        case Map.get(response, "LastEvaluatedKey") do
          nil ->
            {:ok, new_acc}

          last_key ->
            do_query_all_versions_full(aws_client, table, name, last_key, new_acc)
        end

      {:error, {:unexpected_response, %{body: body}}} ->
        parse_aws_error(body, :query)

      {:error, reason} ->
        {:error, {:dynamo_error, :query, reason}}
    end
  end

  defp do_delete_all(_aws_client, _table, [], count) do
    {:ok, count}
  end

  defp do_delete_all(aws_client, table, [version | rest], count) do
    request = %{
      "TableName" => table,
      "Key" => %{
        "name" => %{"S" => version.name},
        "version" => %{"S" => version.version}
      }
    }

    case AWS.DynamoDB.delete_item(aws_client, request) do
      {:ok, _response, _http_response} ->
        do_delete_all(aws_client, table, rest, count + 1)

      {:error, {:unexpected_response, %{body: body}}} ->
        parse_aws_error(body, :delete_item)

      {:error, reason} ->
        {:error, {:dynamo_error, :delete_item, reason}}
    end
  end

  # Response parsing helpers

  defp parse_item(item) do
    %{
      name: get_string(item, "name"),
      version: get_string(item, "version"),
      key: get_string(item, "key"),
      contents: get_string(item, "contents"),
      hmac: get_binary(item, "hmac"),
      digest: get_string(item, "digest"),
      comment: get_string(item, "comment")
    }
  end

  defp parse_list_item(item) do
    %{
      name: get_string(item, "name"),
      version: get_string(item, "version"),
      comment: get_string(item, "comment")
    }
  end

  defp get_string(item, key) do
    case item[key] do
      %{"S" => value} -> value
      _ -> nil
    end
  end

  # Get binary value - HMAC is stored as Binary type (B)
  # The value in DynamoDB is Base64-encoded, decode to get the original hex string
  defp get_binary(item, key) do
    case item[key] do
      %{"B" => base64_value} ->
        case Base.decode64(base64_value) do
          {:ok, decoded} -> decoded
          :error -> nil
        end

      _ ->
        nil
    end
  end

  # Error handling helpers

  defp is_resource_not_found?(body) when is_binary(body) do
    case Jason.decode(body) do
      {:ok, map} -> is_resource_not_found?(map)
      {:error, _} -> false
    end
  end

  defp is_resource_not_found?(map) when is_map(map) do
    error_type = Map.get(map, "__type", "")
    String.contains?(error_type, "ResourceNotFoundException")
  end

  defp is_resource_not_found?(_), do: false

  defp is_conditional_check_failed?(body) when is_binary(body) do
    case Jason.decode(body) do
      {:ok, map} -> is_conditional_check_failed?(map)
      {:error, _} -> false
    end
  end

  defp is_conditional_check_failed?(map) when is_map(map) do
    error_type = Map.get(map, "__type", "")
    String.contains?(error_type, "ConditionalCheckFailedException")
  end

  defp is_conditional_check_failed?(_), do: false

  defp parse_aws_error(body, operation) when is_binary(body) do
    case Jason.decode(body) do
      {:ok, error_map} ->
        parse_aws_error(error_map, operation)

      {:error, _} ->
        {:error, {:dynamo_error, operation, body}}
    end
  end

  defp parse_aws_error(error_map, operation) when is_map(error_map) do
    error_type = Map.get(error_map, "__type", "")
    message = Map.get(error_map, "message") || Map.get(error_map, "Message", "Unknown error")

    error_atom =
      cond do
        String.contains?(error_type, "ResourceNotFoundException") ->
          :table_not_found

        String.contains?(error_type, "ConditionalCheckFailedException") ->
          :already_exists

        String.contains?(error_type, "ValidationException") ->
          :validation_error

        String.contains?(error_type, "AccessDeniedException") ->
          :access_denied

        String.contains?(error_type, "ResourceInUseException") ->
          :resource_in_use

        String.contains?(error_type, "ProvisionedThroughputExceededException") ->
          :throughput_exceeded

        String.contains?(error_type, "ItemCollectionSizeLimitExceededException") ->
          :item_collection_size_exceeded

        String.contains?(error_type, "InternalServerError") ->
          :internal_server_error

        true ->
          :unknown_error
      end

    {:error, {error_atom, operation, message}}
  end

  defp parse_aws_error(error, operation) do
    {:error, {:dynamo_error, operation, error}}
  end
end