lean_lmdb

defmodule LeanLmdb do
  @moduledoc """
  A lean, binary-first Elixir wrapper for LMDB.

  Environments are shared by canonical path within one BEAM OS process. Opaque
  environment and immutable named-database handles may live across calls;
  transactions and cursors never do. Keys and values are raw binaries and read
  results are copied into BEAM-owned binaries.

  ## Example

      iex> {name, version} = LeanLmdb.native_version()
      iex> {name, Version.parse!(version).major}
      {"lean_lmdb", 0}
  """

  alias LeanLmdb.{Batch, Codec, CodecDatabase, Database, Environment, Native}

  @default_map_size 64 * 1024 * 1024
  @default_max_dbs 16
  @default_max_readers 126
  @default_durability :sync
  @minimum_map_size 1024 * 1024
  @maximum_map_size 1024 * 1024 * 1024 * 1024
  @maximum_max_dbs 1024
  @maximum_max_readers 4096
  @option_keys [
    :map_size,
    :max_dbs,
    :max_readers,
    :read_only,
    :create,
    :durability,
    :fixed_map,
    :read_ahead
  ]
  @maximum_key_size 511
  @default_scan_limit 100
  @maximum_scan_limit 10_000
  @default_scan_max_bytes 1024 * 1024
  @maximum_scan_max_bytes 64 * 1024 * 1024
  @maximum_scan_token_bytes 4096
  @scan_option_keys [
    :prefix,
    :from,
    :to,
    :from_inclusive,
    :to_inclusive,
    :limit,
    :max_bytes,
    :continuation
  ]
  @range_scan_option_keys [:from, :to, :from_inclusive, :to_inclusive]

  @typedoc "A stable public error reason."
  @type reason ::
          :database_create_failed
          | :database_not_found
          | :database_open_failed
          | :environment_open_failed
          | :incompatible_environment_options
          | :invalid_database
          | :invalid_database_name
          | :invalid_environment
          | :invalid_key
          | :invalid_value
          | :io_error
          | :database_error
          | :map_full
          | :map_resized
          | :readers_full
          | :transaction_full
          | :out_of_memory
          | :invalid_environment_options
          | :invalid_batch
          | :batch_too_large
          | :mixed_environments
          | :invalid_expected
          | :invalid_replacement
          | :invalid_mode
          | :invalid_limit
          | :invalid_max_bytes
          | :row_too_large
          | :invalid_continuation
          | :stale_continuation
          | :invalid_path
          | :path_not_found
          | :read_only
          | :resource_id_exhausted
          | :transaction_failed
          | :sync_failed
          | :invalid_value_codec
          | :codec_encode_failed
          | :codec_decode_failed
          | :codec_mismatch
          | :unsupported_codec_version
          | :non_deterministic_codec
          | :invalid_batch_writer_options
          | :batch_writer_overloaded
          | :batch_writer_stopping
          | :batch_writer_worker_failed
          | :batch_writer_unknown_outcome
          | {:invalid_option, atom()}

  @typedoc "An ordered operation accepted by `write_batch/2`."
  @type batch_operation ::
          {:put, Database.t(), binary(), binary()}
          | {:put, CodecDatabase.t(), binary(), term()}
          | {:delete, Database.t() | CodecDatabase.t(), binary()}

  @typedoc "The expected state accepted by `compare_exchange/4`."
  @type compare_expected :: :missing | binary()

  @typedoc "The replacement accepted by `compare_exchange/4`."
  @type compare_replacement :: :delete | {:put, binary()}

  @typedoc "An option accepted by `scan/2`."
  @type scan_option ::
          {:prefix, binary()}
          | {:from, binary() | nil}
          | {:to, binary() | nil}
          | {:from_inclusive, boolean()}
          | {:to_inclusive, boolean()}
          | {:limit, 1..10_000}
          | {:max_bytes, 1..67_108_864}
          | {:continuation, binary() | nil}

  @typedoc "LMDB synchronization behavior selected when an environment is opened."
  @type durability :: :sync | :no_meta_sync | :no_sync

  @typedoc "Validated options accepted by `open/2`."
  @type open_option ::
          {:map_size, pos_integer()}
          | {:max_dbs, pos_integer()}
          | {:max_readers, pos_integer()}
          | {:read_only, boolean()}
          | {:create, boolean()}
          | {:durability, durability()}
          | {:fixed_map, boolean()}
          | {:read_ahead, boolean()}

  @doc "Returns the tuple `{\"lean_lmdb\", version}` without accessing LMDB or the filesystem."
  @spec native_version() :: {String.t(), String.t()}
  defdelegate native_version(), to: Native

  @doc """
  Opens an LMDB environment, sharing native state for its canonical path.

  The path is canonicalized after optional creation, so relative components and
  symlinks resolve to the same identity. Defaults are a 64 MiB map, 16 named
  databases, 126 readers, writable access, `create: true`, `durability: :sync`,
  `fixed_map: false`, and `read_ahead: true`.

  `map_size` must be between 1 MiB and 1 TiB and divisible by the operating
  system page size; `max_dbs` is limited to 1..1024 and `max_readers` to
  1..4096. A read-only environment cannot be created.

  `durability: :sync` uses LMDB's fully synchronized default. `:no_meta_sync`
  preserves database integrity but a system crash may undo the latest committed
  transaction. `:no_sync` omits commit-time buffer flushes; a system crash can
  lose transactions and, on a filesystem that reorders writes, can corrupt the
  derived database. Call `sync/1` to force pending buffers to durable storage.

  `fixed_map: true` enables experimental `MDB_FIXEDMAP`. It can fail when the
  recorded virtual address is unavailable and every process opening that
  environment must use compatible settings. It is never enabled implicitly.

  `read_ahead: false` requests LMDB's `MDB_NORDAHEAD` performance hint. It may
  help random reads when the database is larger than RAM, can hurt sequential
  scans, and has no effect on Windows.

  Repeats must use the same effective map size, limits, read mode, durability,
  fixed-map, and read-ahead settings; `create` only controls path/open existence
  behavior.
  Returns `{:ok, environment}` or `{:error, reason}`. The map never grows
  automatically.
  """
  @spec open(Path.t(), [open_option()]) :: {:ok, Environment.t()} | {:error, reason()}
  def open(path, options \\ []) do
    with :ok <- validate_path(path),
         {:ok, validated} <- validate_options(options),
         {:ok, resource} <- native_environment_open(path, validated),
         {:ok, {_id, canonical_path, _map_size, _max_dbs, _max_readers, _read_only}} <-
           Native.environment_info(resource) do
      {:ok,
       %Environment{
         resource: resource,
         path: canonical_path,
         durability: validated.durability,
         fixed_map: validated.fixed_map,
         read_ahead: validated.read_ahead
       }}
    end
  end

  @doc """
  Forces the environment's pending data and metadata buffers to durable storage.

  This calls LMDB's forced `mdb_env_sync`, even when the environment was opened
  with `durability: :no_sync` or `:no_meta_sync`. It is intended for explicit
  group-commit schedules such as a supervised timer. Writes can continue from
  other processes while the dirty-I/O call runs; on return, LMDB reports that
  all commits visible to that synchronization point have been flushed.

  Returns `:ok` or `{:error, reason}`. Read-only and invalid handles are rejected.
  """
  @spec sync(Environment.t()) :: :ok | {:error, reason()}
  def sync(%Environment{resource: resource}) do
    case Native.environment_sync(resource) do
      {:ok, _unit} -> :ok
      {:error, _reason} = error -> error
    end
  end

  def sync(_invalid), do: {:error, :invalid_environment}

  @doc """
  Creates a named binary database, or reopens it when it already exists.

  Names must contain 1..511 bytes of valid UTF-8 and may not contain NUL.
  Returns `{:ok, database}` or `{:error, reason}`; read-only environments cannot
  create databases. The immutable database handle retains its environment.
  """
  @spec create_database(Environment.t(), binary()) ::
          {:ok, Database.t()} | {:error, reason()}
  def create_database(%Environment{} = environment, name) when is_binary(name) do
    case Native.database_create(environment.resource, name) do
      {:ok, resource} ->
        {:ok, %Database{resource: resource, environment: environment, name: name}}

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

  def create_database(%Environment{}, _name), do: {:error, :invalid_database_name}
  def create_database(_environment, _name), do: {:error, :invalid_environment}

  @doc """
  Opens an existing named binary database.

  Names follow the same UTF-8, length, and NUL restrictions as
  `create_database/2`. Returns `{:ok, database}`, `{:error,
  :database_not_found}`, or another documented lifecycle error.
  """
  @spec open_database(Environment.t(), binary()) ::
          {:ok, Database.t()} | {:error, reason()}
  def open_database(%Environment{} = environment, name) when is_binary(name) do
    case Native.database_open(environment.resource, name) do
      {:ok, resource} ->
        {:ok, %Database{resource: resource, environment: environment, name: name}}

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

  def open_database(%Environment{}, _name), do: {:error, :invalid_database_name}
  def open_database(_environment, _name), do: {:error, :invalid_environment}

  @doc """
  Creates an opt-in value-codec view over a raw database handle.

  The codec module must implement `LeanLmdb.ValueCodec`. Its stable binary ID,
  format version, determinism declaration, callback set, and options are
  validated before an immutable `LeanLmdb.CodecDatabase` is returned. No stored
  data is read or migrated by this function.
  """
  @spec with_value_codec(Database.t(), module(), keyword()) ::
          {:ok, CodecDatabase.t()} | {:error, :invalid_database | :invalid_value_codec}
  def with_value_codec(database, codec, options \\ [])

  def with_value_codec(%Database{} = database, codec, options)
      when is_atom(codec) and is_list(options) do
    with true <- bounded_proper_list?(options, 64),
         true <- Keyword.keyword?(options),
         true <- Code.ensure_loaded?(codec),
         true <- codec_callbacks_exported?(codec),
         {:ok, validated_options} <- validate_codec_options(codec, options),
         {:ok, codec_id} <- codec_metadata(codec, :id, &valid_codec_id?/1),
         {:ok, codec_version} <- codec_metadata(codec, :version, &valid_codec_version?/1),
         {:ok, deterministic} <- codec_metadata(codec, :deterministic?, &is_boolean/1) do
      {:ok,
       %CodecDatabase{
         raw_database: database,
         codec: codec,
         options: validated_options,
         codec_id: codec_id,
         codec_version: codec_version,
         deterministic: deterministic
       }}
    else
      _invalid -> {:error, :invalid_value_codec}
    end
  end

  def with_value_codec(%Database{}, _codec, _options), do: {:error, :invalid_value_codec}
  def with_value_codec(_database, _codec, _options), do: {:error, :invalid_database}

  @doc """
  Returns the raw binary database wrapped by a codec database.

  Raw access deliberately bypasses envelope and codec validation and is intended
  for diagnostics and explicit migrations.
  """
  @spec raw_database(CodecDatabase.t()) :: Database.t() | {:error, :invalid_database}
  def raw_database(%CodecDatabase{raw_database: database}), do: database
  def raw_database(_database), do: {:error, :invalid_database}

  @doc """
  Gets the value for `key`.

  Raw database handles return a copied binary. Codec database handles validate
  the stored envelope and decode a logical term after the native transaction
  ends. Returns `{:ok, value}` when present and `:not_found` when missing. Empty
  raw values are therefore distinct from missing keys. Keys must be non-empty
  binaries no larger than LMDB's runtime maximum key size. Native and codec
  failures return `{:error, reason}`; no zero-copy view escapes the call.
  """
  @spec get(Database.t() | CodecDatabase.t(), binary()) ::
          {:ok, binary() | term()} | :not_found | {:error, reason()}
  def get(%CodecDatabase{raw_database: raw_database} = database, key) when is_binary(key) do
    case get(raw_database, key) do
      {:ok, envelope} -> Codec.decode(database, envelope)
      :not_found -> :not_found
      {:error, _reason} = error -> error
    end
  end

  def get(%CodecDatabase{}, _key), do: {:error, :invalid_key}

  def get(%Database{resource: resource}, key) when is_binary(key) do
    case Native.database_get(resource, key) do
      {:ok, :not_found} -> :not_found
      {:ok, value} when is_binary(value) -> {:ok, value}
      {:error, _reason} = error -> error
    end
  end

  def get(%Database{}, _key), do: {:error, :invalid_key}
  def get(_database, _key), do: {:error, :invalid_database}

  @doc """
  Stores `value` under `key`, overwriting any prior value.

  Raw handles require binary keys and values and perform no serialization.
  Codec handles keep keys unchanged and encode/envelope the logical value in the
  caller before entering the NIF. Returns `:ok` after commit or `{:error,
  reason}` without a partial commit.
  """
  @spec put(Database.t() | CodecDatabase.t(), binary(), term()) :: :ok | {:error, reason()}
  def put(%CodecDatabase{raw_database: raw_database} = database, key, value)
      when is_binary(key) do
    with {:ok, envelope} <- Codec.encode(database, value) do
      put(raw_database, key, envelope)
    end
  end

  def put(%CodecDatabase{}, _key, _value), do: {:error, :invalid_key}

  def put(%Database{resource: resource}, key, value)
      when is_binary(key) and is_binary(value) do
    case Native.database_put(resource, key, value) do
      {:ok, _unit} -> :ok
      {:error, _reason} = error -> error
    end
  end

  def put(%Database{}, key, _value) when not is_binary(key), do: {:error, :invalid_key}
  def put(%Database{}, _key, _value), do: {:error, :invalid_value}
  def put(_database, _key, _value), do: {:error, :invalid_database}

  @doc """
  Deletes `key` if present.

  Deletion is idempotent: both present and missing keys return `:ok`. Validation,
  read-only, and LMDB failures return `{:error, reason}`.
  """
  @spec delete(Database.t() | CodecDatabase.t(), binary()) :: :ok | {:error, reason()}
  def delete(%CodecDatabase{raw_database: raw_database}, key), do: delete(raw_database, key)

  def delete(%Database{resource: resource}, key) when is_binary(key) do
    case Native.database_delete(resource, key) do
      {:ok, _unit} -> :ok
      {:error, _reason} = error -> error
    end
  end

  def delete(%Database{}, _key), do: {:error, :invalid_key}
  def delete(_database, _key), do: {:error, :invalid_database}

  @doc """
  Returns one bounded page of key/value pairs in ascending binary-key order.

  Raw handles return copied binary values. Codec handles decode copied envelopes
  after the native scan ends. `max_bytes` always counts encoded key/value bytes,
  not decoded BEAM heap size; codec implementations require independent decoded
  expansion limits.

  A scan is either a prefix scan (`prefix: binary`) or an explicit range scan.
  Range endpoints default to `nil`; the lower endpoint is inclusive and the
  upper endpoint exclusive by default. Prefix cannot be mixed with `from`, `to`,
  or the inclusivity options. Prefixes and endpoints are at most 511 bytes. `limit` defaults to 100 (maximum 10,000) and
  `max_bytes` defaults to 1 MiB (maximum 64 MiB), counting key plus value bytes.

  The third result element is `:done` or an opaque continuation binary. Pass the
  continuation with the same database and bounds to resume exclusively after
  the last emitted key. Page limits may change while resuming. If the
  first eligible row exceeds `max_bytes`, the call returns `{:error,
  :row_too_large}` and no rows, making the output byte budget a hard bound. Invalid options/tokens and
  native read failures also return `{:error, reason}`. Pages do not share a
  snapshot and all row/token bytes are copied before the call returns.
  """
  @spec scan(Database.t() | CodecDatabase.t(), [scan_option()]) ::
          {:ok, [{binary(), term()}], :done | binary()} | {:error, reason()}
  def scan(database, options \\ [])

  def scan(%CodecDatabase{raw_database: raw_database} = database, options) do
    with {:ok, rows, continuation} <- scan(raw_database, options),
         {:ok, decoded_rows} <- decode_codec_rows(database, rows) do
      {:ok, decoded_rows, continuation}
    end
  end

  def scan(%Database{resource: resource}, options) do
    with {:ok, mode, limit, max_bytes, continuation} <- validate_scan_options(options) do
      case Native.database_scan(resource, mode, limit, max_bytes, continuation) do
        {:ok, {rows, next}} when is_list(rows) and (next == :done or is_binary(next)) ->
          {:ok, rows, next}

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

  def scan(_database, _options), do: {:error, :invalid_database}

  @doc """
  Applies an ordered list of puts and deletes atomically across named databases.

  Every database must belong to exactly the supplied environment. The complete
  list is validated before one write transaction is opened, operations are
  applied in list order, and missing deletes are idempotent. Empty batches are
  valid no-ops. A batch is limited to 10,000 operations and 64 MiB of aggregate
  encoded key and value bytes. All codec values are encoded before the native
  call, so a callback failure leaves the complete mixed raw/codec batch
  unexecuted. Returns `:ok` only after commit; validation, mixed environments,
  read-only access, and LMDB failures return `{:error, reason}` with the
  transaction aborted.
  """
  @spec write_batch(Environment.t(), [batch_operation()]) :: :ok | {:error, reason()}
  def write_batch(%Environment{} = environment, operations) do
    with {:ok, prepared} <- Batch.prepare(environment, operations) do
      Batch.commit(environment, prepared)
    end
  end

  def write_batch(_environment, _operations), do: {:error, :invalid_environment}

  @doc """
  Atomically compares a key's current state and applies a replacement on match.

  `:missing` is distinct from an expected empty binary. A mismatch returns the
  exact current state as `{:conflict, :missing}` or `{:conflict, binary}` and
  never mutates the database. Missing deletes after a successful comparison are
  idempotent. A match returns `:ok`; malformed arguments and native failures
  return `{:error, reason}`. Conflict binaries are copied before return.

  Codec handles compare logical values by deterministic encoded envelope bytes
  and decode conflicts. Non-deterministic codecs return
  `{:error, :non_deterministic_codec}`; codec versions/options must match the
  representation already stored.
  """
  @spec compare_exchange(
          Database.t() | CodecDatabase.t(),
          binary(),
          compare_expected() | term(),
          compare_replacement() | {:put, term()}
        ) :: :ok | {:conflict, :missing | term()} | {:error, reason()}
  def compare_exchange(%CodecDatabase{deterministic: false}, _key, _expected, _replacement),
    do: {:error, :non_deterministic_codec}

  def compare_exchange(
        %CodecDatabase{raw_database: raw_database} = database,
        key,
        expected,
        replacement
      )
      when is_binary(key) do
    with :ok <- validate_mutation_key(key),
         {:ok, encoded_expected} <- encode_codec_expected(database, expected),
         {:ok, encoded_replacement} <- encode_codec_replacement(database, replacement) do
      raw_database
      |> compare_exchange(key, encoded_expected, encoded_replacement)
      |> decode_codec_compare_result(database)
    end
  end

  def compare_exchange(%CodecDatabase{}, _key, _expected, _replacement),
    do: {:error, :invalid_key}

  def compare_exchange(%Database{resource: resource}, key, expected, replacement)
      when is_binary(key) do
    with :ok <- validate_mutation_key(key),
         :ok <- validate_expected(expected),
         :ok <- validate_replacement(replacement) do
      case Native.database_compare_exchange(resource, key, expected, replacement) do
        {:ok, :ok} -> :ok
        {:ok, {:conflict, current}} -> {:conflict, current}
        {:error, _reason} = error -> error
      end
    end
  end

  def compare_exchange(%Database{}, _key, _expected, _replacement),
    do: {:error, :invalid_key}

  def compare_exchange(_database, _key, _expected, _replacement),
    do: {:error, :invalid_database}

  @doc """
  Returns deterministic lifecycle information for an environment or database.

  Environment maps contain `:resource_id`, canonical `:path`, `:map_size`,
  `:max_dbs`, `:max_readers`, `:read_only`, `:durability`, `:fixed_map`, and
  `:read_ahead`; database maps additionally contain `:name`. Invalid handles return
  `{:error, :invalid_environment}`.
  """
  @spec info(Environment.t() | Database.t() | CodecDatabase.t()) ::
          map() | {:error, :invalid_environment}
  def info(%Environment{
        resource: resource,
        durability: durability,
        fixed_map: fixed_map,
        read_ahead: read_ahead
      }) do
    case Native.environment_info(resource) do
      {:ok, {id, path, map_size, max_dbs, max_readers, read_only}} ->
        %{
          resource_id: id,
          path: path,
          map_size: map_size,
          max_dbs: max_dbs,
          max_readers: max_readers,
          read_only: read_only,
          durability: durability,
          fixed_map: fixed_map,
          read_ahead: read_ahead
        }

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

  def info(%CodecDatabase{raw_database: raw_database} = database) do
    case info(raw_database) do
      %{} = raw_info ->
        Map.put(raw_info, :value_codec, %{
          id: database.codec_id,
          version: database.codec_version,
          deterministic: database.deterministic
        })

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

  def info(%Database{resource: resource, environment: environment}) do
    with {:ok, {resource_id, name}} <- Native.database_info(resource),
         %{resource_id: ^resource_id} = environment_info <- info(environment) do
      Map.put(environment_info, :name, name)
    else
      _invalid -> {:error, :invalid_environment}
    end
  end

  def info(_invalid), do: {:error, :invalid_environment}

  @doc """
  Checks the LMDB reader table and clears slots left by terminated processes.

  Returns `{:ok, cleared_count}`. Active readers are never cleared. This is
  backed by heed's `Env::clear_stale_readers` and may be called while the
  environment is in use. Invalid handles and native failures return `{:error,
  reason}`.
  """
  @spec clear_stale_readers(Environment.t()) ::
          {:ok, non_neg_integer()} | {:error, reason()}
  def clear_stale_readers(%Environment{resource: resource}) do
    Native.environment_clear_stale_readers(resource)
  end

  def clear_stale_readers(_invalid), do: {:error, :invalid_environment}

  @doc """
  Returns process-local registry counts and prunes stale weak entries.

  This exposes counts only; it never reveals native pointers. It is intended for
  deterministic lifecycle diagnostics and tests. The returned map always has
  non-negative `:live`, `:stale`, and `:total` counts.
  """
  @spec registry_info() :: %{
          live: non_neg_integer(),
          stale: non_neg_integer(),
          total: non_neg_integer()
        }
  def registry_info do
    {live, stale, total} = Native.registry_info()
    %{live: live, stale: stale, total: total}
  end

  defp validate_scan_options(options) when is_list(options) do
    with :ok <- validate_scan_keyword(options),
         values = Map.new(options),
         {:ok, mode} <- validate_scan_mode(values),
         limit = Map.get(values, :limit, @default_scan_limit),
         max_bytes = Map.get(values, :max_bytes, @default_scan_max_bytes),
         continuation = Map.get(values, :continuation),
         :ok <- validate_scan_limit(limit),
         :ok <- validate_scan_max_bytes(max_bytes),
         :ok <- validate_scan_continuation(continuation) do
      {:ok, mode, limit, max_bytes, continuation}
    end
  end

  defp validate_scan_options(_options), do: {:error, {:invalid_option, :options}}

  defp validate_scan_keyword(options) do
    if bounded_proper_list?(options, length(@scan_option_keys)) and Keyword.keyword?(options) do
      keys = Keyword.keys(options)

      cond do
        length(keys) != length(Enum.uniq(keys)) ->
          {:error, {:invalid_option, :options}}

        invalid_key = Enum.find(keys, &(&1 not in @scan_option_keys)) ->
          {:error, {:invalid_option, invalid_key}}

        :prefix in keys and Enum.any?(@range_scan_option_keys, &(&1 in keys)) ->
          {:error, {:invalid_option, :options}}

        true ->
          :ok
      end
    else
      {:error, {:invalid_option, :options}}
    end
  end

  defp validate_scan_mode(%{prefix: prefix})
       when is_binary(prefix) and byte_size(prefix) <= @maximum_key_size,
       do: {:ok, {:prefix, prefix}}

  defp validate_scan_mode(%{prefix: _prefix}), do: {:error, {:invalid_option, :prefix}}

  defp validate_scan_mode(values) do
    from = Map.get(values, :from)
    to = Map.get(values, :to)
    from_inclusive = Map.get(values, :from_inclusive, true)
    to_inclusive = Map.get(values, :to_inclusive, false)

    with :ok <- validate_optional_binary(:from, from),
         :ok <- validate_optional_binary(:to, to),
         :ok <- validate_boolean(:from_inclusive, from_inclusive),
         :ok <- validate_boolean(:to_inclusive, to_inclusive) do
      {:ok, {:range, from, to, from_inclusive, to_inclusive}}
    end
  end

  defp validate_optional_binary(_key, nil), do: :ok

  defp validate_optional_binary(_key, value)
       when is_binary(value) and byte_size(value) <= @maximum_key_size,
       do: :ok

  defp validate_optional_binary(key, _value), do: {:error, {:invalid_option, key}}

  defp validate_scan_limit(limit)
       when is_integer(limit) and limit >= 1 and limit <= @maximum_scan_limit,
       do: :ok

  defp validate_scan_limit(_limit), do: {:error, {:invalid_option, :limit}}

  defp validate_scan_max_bytes(max_bytes)
       when is_integer(max_bytes) and max_bytes >= 1 and max_bytes <= @maximum_scan_max_bytes,
       do: :ok

  defp validate_scan_max_bytes(_max_bytes), do: {:error, {:invalid_option, :max_bytes}}

  defp validate_scan_continuation(nil), do: :ok

  defp validate_scan_continuation(continuation)
       when is_binary(continuation) and byte_size(continuation) <= @maximum_scan_token_bytes,
       do: :ok

  defp validate_scan_continuation(_continuation),
    do: {:error, {:invalid_option, :continuation}}

  defp validate_mutation_key(key)
       when byte_size(key) >= 1 and byte_size(key) <= @maximum_key_size,
       do: :ok

  defp validate_mutation_key(_key), do: {:error, :invalid_key}

  defp decode_codec_compare_result({:conflict, current}, database) when is_binary(current) do
    case Codec.decode(database, current) do
      {:ok, value} -> {:conflict, value}
      {:error, _reason} = error -> error
    end
  end

  defp decode_codec_compare_result(result, _database), do: result

  defp decode_codec_rows(database, rows) do
    Enum.reduce_while(rows, {:ok, []}, fn {key, envelope}, {:ok, decoded} ->
      case Codec.decode(database, envelope) do
        {:ok, value} -> {:cont, {:ok, [{key, value} | decoded]}}
        {:error, _reason} = error -> {:halt, error}
      end
    end)
    |> case do
      {:ok, decoded} -> {:ok, Enum.reverse(decoded)}
      {:error, _reason} = error -> error
    end
  end

  defp encode_codec_expected(_database, :missing), do: {:ok, :missing}
  defp encode_codec_expected(database, expected), do: Codec.encode(database, expected)

  defp encode_codec_replacement(_database, :delete), do: {:ok, :delete}

  defp encode_codec_replacement(database, {:put, replacement}) do
    case Codec.encode(database, replacement) do
      {:ok, encoded} -> {:ok, {:put, encoded}}
      {:error, _reason} = error -> error
    end
  end

  defp encode_codec_replacement(_database, _replacement), do: {:error, :invalid_replacement}

  defp validate_expected(:missing), do: :ok
  defp validate_expected(expected) when is_binary(expected), do: :ok
  defp validate_expected(_expected), do: {:error, :invalid_expected}

  defp validate_replacement(:delete), do: :ok
  defp validate_replacement({:put, value}) when is_binary(value), do: :ok
  defp validate_replacement(_replacement), do: {:error, :invalid_replacement}

  defp native_environment_open(path, options) do
    Native.environment_open(
      path,
      options.map_size,
      options.max_dbs,
      options.max_readers,
      options.read_only,
      options.create,
      durability_code(options.durability),
      options.fixed_map,
      options.read_ahead
    )
  end

  defp validate_path(path) when is_binary(path) do
    if path != "" and String.valid?(path) and not String.contains?(path, <<0>>) do
      :ok
    else
      {:error, {:invalid_option, :path}}
    end
  end

  defp validate_path(_path), do: {:error, {:invalid_option, :path}}

  defp validate_options(options) when is_list(options) do
    with :ok <- validate_keyword_options(options),
         values = Map.new(options),
         map_size = Map.get(values, :map_size, @default_map_size),
         max_dbs = Map.get(values, :max_dbs, @default_max_dbs),
         max_readers = Map.get(values, :max_readers, @default_max_readers),
         read_only = Map.get(values, :read_only, false),
         create = Map.get(values, :create, true),
         durability = Map.get(values, :durability, @default_durability),
         fixed_map = Map.get(values, :fixed_map, false),
         read_ahead = Map.get(values, :read_ahead, true),
         :ok <- validate_map_size(map_size),
         :ok <- validate_range(:max_dbs, max_dbs, @maximum_max_dbs),
         :ok <- validate_range(:max_readers, max_readers, @maximum_max_readers),
         :ok <- validate_boolean(:read_only, read_only),
         :ok <- validate_boolean(:create, create),
         :ok <- validate_durability(durability),
         :ok <- validate_boolean(:fixed_map, fixed_map),
         :ok <- validate_boolean(:read_ahead, read_ahead),
         :ok <- validate_access_mode(read_only, create, durability) do
      {:ok,
       %{
         map_size: map_size,
         max_dbs: max_dbs,
         max_readers: max_readers,
         read_only: read_only,
         create: create,
         durability: durability,
         fixed_map: fixed_map,
         read_ahead: read_ahead
       }}
    end
  end

  defp validate_options(_options), do: {:error, {:invalid_option, :options}}

  defp validate_keyword_options(options) do
    if bounded_proper_list?(options, length(@option_keys)) and Keyword.keyword?(options) do
      keys = Keyword.keys(options)

      cond do
        length(keys) != length(Enum.uniq(keys)) ->
          {:error, {:invalid_option, :options}}

        invalid_key = Enum.find(keys, &(&1 not in @option_keys)) ->
          {:error, {:invalid_option, invalid_key}}

        true ->
          :ok
      end
    else
      {:error, {:invalid_option, :options}}
    end
  end

  defp validate_map_size(value)
       when is_integer(value) and value >= @minimum_map_size and value <= @maximum_map_size do
    if rem(value, Native.system_page_size()) == 0 do
      :ok
    else
      {:error, {:invalid_option, :map_size}}
    end
  end

  defp validate_map_size(_value), do: {:error, {:invalid_option, :map_size}}

  defp validate_range(_key, value, maximum)
       when is_integer(value) and value >= 1 and value <= maximum,
       do: :ok

  defp validate_range(key, _value, _maximum), do: {:error, {:invalid_option, key}}

  defp bounded_proper_list?([], _remaining), do: true
  defp bounded_proper_list?(_list, 0), do: false

  defp bounded_proper_list?([_head | tail], remaining),
    do: bounded_proper_list?(tail, remaining - 1)

  defp bounded_proper_list?(_improper, _remaining), do: false

  defp validate_boolean(_key, value) when is_boolean(value), do: :ok
  defp validate_boolean(key, _value), do: {:error, {:invalid_option, key}}

  defp validate_durability(value) when value in [:sync, :no_meta_sync, :no_sync], do: :ok
  defp validate_durability(_value), do: {:error, {:invalid_option, :durability}}

  defp validate_access_mode(true, true, _durability), do: {:error, {:invalid_option, :create}}

  defp validate_access_mode(true, _create, durability) when durability != :sync,
    do: {:error, {:invalid_option, :durability}}

  defp validate_access_mode(_read_only, _create, _durability), do: :ok

  defp codec_callbacks_exported?(codec) do
    function_exported?(codec, :id, 0) and
      function_exported?(codec, :version, 0) and
      function_exported?(codec, :deterministic?, 0) and
      function_exported?(codec, :encode, 2) and
      function_exported?(codec, :decode, 3)
  end

  defp validate_codec_options(codec, options) do
    if function_exported?(codec, :validate_options, 1) do
      case safe_codec_apply(codec, :validate_options, [options]) do
        {:ok, {:ok, validated}} -> validate_codec_options_result(validated)
        _invalid -> {:error, :invalid_value_codec}
      end
    else
      {:ok, options}
    end
  end

  defp validate_codec_options_result(validated) when is_list(validated) do
    if bounded_proper_list?(validated, 64) and Keyword.keyword?(validated),
      do: {:ok, validated},
      else: {:error, :invalid_value_codec}
  end

  defp validate_codec_options_result(_validated), do: {:error, :invalid_value_codec}

  defp codec_metadata(codec, callback, validator) do
    case safe_codec_apply(codec, callback, []) do
      {:ok, value} -> if validator.(value), do: {:ok, value}, else: {:error, :invalid_value_codec}
      :error -> {:error, :invalid_value_codec}
    end
  end

  defp safe_codec_apply(codec, callback, arguments) do
    {:ok, apply(codec, callback, arguments)}
  rescue
    _exception -> :error
  catch
    _kind, _reason -> :error
  end

  defp valid_codec_id?(id), do: is_binary(id) and byte_size(id) in 1..255

  defp valid_codec_version?(version),
    do: is_integer(version) and version in 0..4_294_967_295

  defp durability_code(:sync), do: 0
  defp durability_code(:no_meta_sync), do: 1
  defp durability_code(:no_sync), do: 2
end