defmodule Concord do
  @moduledoc """
  Public API for the Concord distributed key-value store.

  Concord is a CP (Consistent + Partition-tolerant) distributed KV store
  built on the Raft consensus algorithm via the `ra` library.

  ## Examples

      iex> Concord.put("user:123", %{name: "Alice"})
      :ok

      iex> Concord.get("user:123")
      {:ok, %{name: "Alice"}}

      iex> Concord.delete("user:123")
      :ok
  """

  require Logger

  alias Concord.{Compression, StateMachine, TTL, Txn}

  @timeout 5_000
  @cluster_name :concord_cluster

  @doc """
  Stores a key-value pair in the cluster.

  Values are automatically compressed if they exceed the configured size threshold
  (default: 1KB). Compression is transparent and values are automatically
  decompressed when retrieved.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:ttl` - Time-to-live in seconds (default: nil for no expiration)
  - `:compress` - Override automatic compression (true/false)
  """
  def put(key, value, opts \\ []) do
    with :ok <- validate_key(key),
         :ok <- validate_ttl_option(opts) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      ttl_option = Keyword.get(opts, :ttl)
      expires_at = calculate_expires_at(ttl_option)
      start_time = System.monotonic_time()

      # Apply compression if enabled and value is large enough
      compressed_value = maybe_compress(value, opts)
      was_compressed = compressed_value != value

      result =
        case command({:put, key, compressed_value, expires_at}, timeout) do
          {:ok, :ok, _} -> :ok
          {:ok, result, _} -> {:ok, result}
          {:timeout, _} -> {:error, :timeout}
          {:error, :noproc} -> {:error, :cluster_not_ready}
          {:error, reason} -> {:error, reason}
        end

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :put],
        %{duration: duration},
        %{result: result, has_ttl: ttl_option != nil, compressed: was_compressed}
      )

      result
    end
  end

  @doc """
  Retrieves a value by key from the cluster.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)
    - `:eventual` - Fastest, may return stale data (reads from any node)
    - `:leader` - Balanced, reads from leader node
    - `:strong` - Linearizable reads with heartbeat verification (slowest)
  """
  def get(key, opts \\ []) do
    with :ok <- validate_key(key) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      consistency = Keyword.get(opts, :consistency, default_consistency())
      start_time = System.monotonic_time()

      result =
        case query({:get, key}, timeout, consistency) do
          {:ok, {{_index, _term}, query_result}, _} ->
            query_result

          {:timeout, _} ->
            {:error, :timeout}

          {:error, :noproc} ->
            {:error, :cluster_not_ready}

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

      # Automatically decompress the result if needed
      decompressed_result =
        case result do
          {:ok, value} -> {:ok, Compression.decompress(value)}
          other -> other
        end

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :get],
        %{duration: duration},
        %{result: decompressed_result, consistency: consistency}
      )

      decompressed_result
    end
  end

  @doc """
  Deletes a key from the cluster.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  """
  def delete(key, opts \\ []) do
    with :ok <- validate_key(key) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      start_time = System.monotonic_time()

      result =
        case command({:delete, key}, timeout) do
          {:ok, :ok, _} -> :ok
          {:ok, result, _} -> {:ok, result}
          {:timeout, _} -> {:error, :timeout}
          {:error, :noproc} -> {:error, :cluster_not_ready}
          {:error, reason} -> {:error, reason}
        end

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :delete],
        %{duration: duration},
        %{result: result}
      )

      result
    end
  end

  @doc """
  Conditionally updates a key only if it matches the expected value (compare-and-swap).

  This provides atomic conditional updates, useful for implementing locks, counters,
  and other concurrent data structures.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:ttl` - Time-to-live in seconds for the new value
  - `:compress` - Override automatic compression (true/false)

  ## Examples

      # Only update if current value matches
      Concord.put_if("counter", 1, expected: 0)
      # => :ok

      # Fails if value doesn't match
      Concord.put_if("counter", 2, expected: 0)
      # => {:error, :condition_failed}

      # Conditional update with predicate
      Concord.put_if("config", new_config,
        condition: fn old_value -> old_value.version < new_config.version end
      )
  """
  def put_if(key, value, opts) do
    with :ok <- validate_key(key),
         :ok <- validate_ttl_option(opts),
         :ok <- validate_condition_opts(opts) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      ttl_option = Keyword.get(opts, :ttl)
      expires_at = calculate_expires_at(ttl_option)
      compressed_value = maybe_compress(value, opts)

      expected = Keyword.get(opts, :expected)
      condition_fn = Keyword.get(opts, :condition)

      start_time = System.monotonic_time()

      # If a condition function is provided, evaluate it PRE-CONSENSUS
      # and convert to a CAS (compare-and-swap) operation with :expected.
      # This keeps anonymous functions out of the Raft log.
      result =
        cond do
          expected != nil ->
            unwrap_command_result({:put_if, key, compressed_value, expires_at, expected}, timeout)

          condition_fn != nil ->
            evaluate_condition_then_cas(
              key,
              fn current -> {:put_if, key, compressed_value, expires_at, current} end,
              condition_fn,
              opts,
              timeout
            )

          true ->
            {:error, :missing_condition}
        end

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :put_if],
        %{duration: duration},
        %{result: result, has_condition: condition_fn != nil, has_expected: expected != nil}
      )

      result
    end
  end

  @doc """
  Conditionally deletes a key only if it matches the expected value.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples

      # Only delete if current value matches
      Concord.delete_if("lock", expected: "my_lock_id")
      # => :ok

      # Fails if value doesn't match
      Concord.delete_if("lock", expected: "different_id")
      # => {:error, :condition_failed}

      # Delete with predicate
      Concord.delete_if("temp:file",
        condition: fn value -> value.created_at < cutoff_time end
      )
  """
  def delete_if(key, opts) do
    with :ok <- validate_key(key),
         :ok <- validate_condition_opts(opts) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      expected = Keyword.get(opts, :expected)
      condition_fn = Keyword.get(opts, :condition)

      start_time = System.monotonic_time()

      result =
        cond do
          expected != nil ->
            unwrap_command_result({:delete_if, key, expected, nil}, timeout)

          condition_fn != nil ->
            evaluate_condition_then_cas(
              key,
              fn current -> {:delete_if, key, current, nil} end,
              condition_fn,
              opts,
              timeout
            )

          true ->
            {:error, :missing_condition}
        end

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :delete_if],
        %{duration: duration},
        %{result: result, has_condition: condition_fn != nil, has_expected: expected != nil}
      )

      result
    end
  end

  @doc """
  Returns all key-value pairs in the store.
  Use sparingly on large datasets.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)
  """
  def get_all(opts \\ []) do
    timeout = Keyword.get(opts, :timeout, @timeout)
    consistency = Keyword.get(opts, :consistency, default_consistency())

    case query(:get_all, timeout, consistency) do
      {:ok, {{_index, _term}, query_result}, _} -> query_result
      {:timeout, _} -> {:error, :timeout}
      {:error, :noproc} -> {:error, :cluster_not_ready}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc """
  Returns cluster status information.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)
  """
  def status(opts \\ []) do
    timeout = Keyword.get(opts, :timeout, @timeout)
    consistency = Keyword.get(opts, :consistency, default_consistency())
    server_id = server_id()

    with {:ok, overview, _} <- :ra.member_overview(server_id, timeout),
         {:ok, {{_index, _term}, {:ok, stats}}, _} <- query(:stats, timeout, consistency) do
      # Convert overview to JSON-friendly format (recursively convert tuples to strings)
      json_friendly_overview = make_json_friendly(overview)

      {:ok,
       %{
         cluster: json_friendly_overview,
         storage: stats,
         node: node()
       }}
    else
      {:error, reason} -> {:error, reason}
      {:timeout, _} -> {:error, :timeout}
    end
  end

  @doc """
  Returns cluster members information.
  """
  def members do
    server_id = server_id()

    case :ra.members(server_id) do
      {:ok, members, _leader} -> {:ok, members}
      {:error, reason} -> {:error, reason}
      {:timeout, _} -> {:error, :timeout}
    end
  end

  @doc """
  Stores a key-value pair with an explicit TTL in seconds.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples
      iex> Concord.put_with_ttl("cache:user:123", %{data: "value"}, 3600)
      :ok
  """
  def put_with_ttl(key, value, ttl_seconds, opts \\ [])
      when is_integer(ttl_seconds) and ttl_seconds > 0 do
    put(key, value, Keyword.put(opts, :ttl, ttl_seconds))
  end

  @doc """
  Extends the TTL of an existing key by the specified number of seconds.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples
      iex> Concord.touch("cache:user:123", 1800)
      :ok
  """
  def touch(key, additional_ttl_seconds, opts \\ [])
      when is_integer(additional_ttl_seconds) and additional_ttl_seconds > 0 do
    with :ok <- validate_key(key) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      start_time = System.monotonic_time()

      result =
        case command({:touch, key, additional_ttl_seconds}, timeout) do
          {:ok, :ok, _} -> :ok
          {:ok, result, _} -> result
          {:timeout, _} -> {:error, :timeout}
          {:error, :noproc} -> {:error, :cluster_not_ready}
          {:error, reason} -> {:error, reason}
        end

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :touch],
        %{duration: duration},
        %{result: result, key: key}
      )

      result
    end
  end

  @doc """
  Gets the remaining TTL for a key in seconds.

  Returns nil if the key has no expiration.
  Returns {:error, :not_found} if the key doesn't exist or has expired.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)

  ## Examples
      iex> Concord.ttl("cache:user:123")
      {:ok, 1800}
  """
  def ttl(key, opts \\ []) do
    with :ok <- validate_key(key) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      consistency = Keyword.get(opts, :consistency, default_consistency())
      start_time = System.monotonic_time()

      result =
        case query({:ttl, key}, timeout, consistency) do
          {:ok, {{_index, _term}, query_result}, _} ->
            query_result

          {:timeout, _} ->
            {:error, :timeout}

          {:error, :noproc} ->
            {:error, :cluster_not_ready}

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

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :ttl],
        %{duration: duration},
        %{result: result, key: key, consistency: consistency}
      )

      result
    end
  end

  @doc """
  Gets a value along with its remaining TTL.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)

  ## Examples
      iex> Concord.get_with_ttl("cache:user:123")
      {:ok, {%{data: "value"}, 1800}}
  """
  def get_with_ttl(key, opts \\ []) do
    with :ok <- validate_key(key) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      consistency = Keyword.get(opts, :consistency, default_consistency())
      start_time = System.monotonic_time()

      result =
        case query({:get_with_ttl, key}, timeout, consistency) do
          {:ok, {{_index, _term}, query_result}, _} ->
            query_result

          {:timeout, _} ->
            {:error, :timeout}

          {:error, :noproc} ->
            {:error, :cluster_not_ready}

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

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :get_with_ttl],
        %{duration: duration},
        %{result: result, key: key, consistency: consistency}
      )

      result
    end
  end

  @doc """
  Returns all key-value pairs with their TTL information.
  Use sparingly on large datasets.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)

  ## Examples
      iex> Concord.get_all_with_ttl(token: "token")
      {:ok, %{"key1" => %{value: "val1", ttl: 3600}, "key2" => %{value: "val2", ttl: nil}}}
  """
  def get_all_with_ttl(opts \\ []) do
    timeout = Keyword.get(opts, :timeout, @timeout)
    consistency = Keyword.get(opts, :consistency, default_consistency())

    case query(:get_all_with_ttl, timeout, consistency) do
      {:ok, {{_index, _term}, query_result}, _} -> query_result
      {:timeout, _} -> {:error, :timeout}
      {:error, :noproc} -> {:error, :cluster_not_ready}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc """
  Returns key-value pairs for all keys starting with the given prefix.

  Uses an efficient server-side scan on the ordered ETS table, avoiding
  loading all keys into memory. O(log N + K) where K is the number of
  matching keys.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)

  ## Examples
      iex> Concord.prefix_scan("user:")
      {:ok, [{"user:1", %{name: "Alice"}}, {"user:2", %{name: "Bob"}}]}
  """
  def prefix_scan(prefix, opts \\ []) when is_binary(prefix) do
    timeout = Keyword.get(opts, :timeout, @timeout)
    consistency = Keyword.get(opts, :consistency, default_consistency())

    case query({:prefix_scan, prefix}, timeout, consistency) do
      {:ok, {{_index, _term}, {:ok, pairs}}, _} -> {:ok, decompress_pairs(pairs)}
      {:ok, {{_index, _term}, query_result}, _} -> query_result
      {:timeout, _} -> {:error, :timeout}
      {:error, :noproc} -> {:error, :cluster_not_ready}
      {:error, reason} -> {:error, reason}
    end
  end

  @doc """
  Stores multiple key-value pairs in the cluster atomically.

  Either all operations succeed or all fail together.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples
      iex> Concord.put_many([{"key1", "value1"}, {"key2", "value2"}])
      {:ok, %{"key1" => :ok, "key2" => :ok}}
  """
  def put_many(operations, opts \\ []) when is_list(operations) do
    with :ok <- validate_batch_size(operations),
         :ok <- validate_put_operations(operations) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      start_time = System.monotonic_time()

      # Convert operations to expected format for StateMachine
      formatted_operations =
        Enum.map(operations, fn
          {key, value} -> {key, value, nil}
          {key, value, ttl} when is_integer(ttl) -> {key, value, calculate_expires_at(ttl)}
          {key, value, expires_at} -> {key, value, expires_at}
        end)

      result = process_batch_command_result(command({:put_many, formatted_operations}, timeout))

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :put_many],
        %{duration: duration},
        %{result: result, batch_size: length(operations)}
      )

      result
    end
  end

  @doc """
  Stores multiple key-value pairs with a TTL atomically.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples
      iex> Concord.put_many_with_ttl([{"key1", "value1"}, {"key2", "value2"}], 3600)
      {:ok, %{"key1" => :ok, "key2" => :ok}}
  """
  def put_many_with_ttl(operations, ttl_seconds, opts \\ [])
      when is_list(operations) and is_integer(ttl_seconds) and ttl_seconds > 0 do
    # Add TTL to each operation
    operations_with_ttl =
      Enum.map(operations, fn {key, value} ->
        {key, value, ttl_seconds}
      end)

    put_many(operations_with_ttl, opts)
  end

  @doc """
  Retrieves multiple values by key from the cluster.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:consistency` - Read consistency level (default: :leader)

  ## Examples
      iex> Concord.get_many(["key1", "key2"])
      {:ok, %{"key1" => {:ok, "value1"}, "key2" => {:error, :not_found}}}
  """
  def get_many(keys, opts \\ []) when is_list(keys) do
    with :ok <- validate_batch_size(keys),
         :ok <- validate_keys(keys) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      consistency = Keyword.get(opts, :consistency, default_consistency())
      start_time = System.monotonic_time()

      # Use the query path (not a write command) for reads
      result =
        case query({:get_many, keys}, timeout, consistency) do
          {:ok, {{_index, _term}, query_result}, _} ->
            query_result

          {:timeout, _} ->
            {:error, :timeout}

          {:error, :noproc} ->
            {:error, :cluster_not_ready}

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

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :get_many],
        %{duration: duration},
        %{result: result, batch_size: length(keys), consistency: consistency}
      )

      result
    end
  end

  @doc """
  Deletes multiple keys from the cluster atomically.

  Either all operations succeed or all fail together.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples
      iex> Concord.delete_many(["key1", "key2"])
      {:ok, %{"key1" => :ok, "key2" => :ok}}
  """
  def delete_many(keys, opts \\ []) when is_list(keys) do
    with :ok <- validate_batch_size(keys),
         :ok <- validate_keys(keys) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      start_time = System.monotonic_time()

      result = process_batch_command_result(command({:delete_many, keys}, timeout))

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :delete_many],
        %{duration: duration},
        %{result: result, batch_size: length(keys)}
      )

      result
    end
  end

  @doc """
  Extends the TTL of multiple keys atomically.

  Either all operations succeed or all fail together.

  ## Options
  - `:timeout` - Operation timeout in milliseconds (default: 5000)
  - `:token` - Authentication token (required if auth is enabled)

  ## Examples
      iex> Concord.touch_many([{"key1", 1800}, {"key2", 3600}])
      {:ok, %{"key1" => :ok, "key2" => :ok}}
  """
  def touch_many(operations, opts \\ []) when is_list(operations) do
    with :ok <- validate_batch_size(operations),
         :ok <- validate_touch_operations(operations) do
      timeout = Keyword.get(opts, :timeout, @timeout)
      start_time = System.monotonic_time()

      result =
        case command({:touch_many, operations}, timeout) do
          {:ok, {:ok, results}, _} ->
            # Convert StateMachine results to map format
            result_map = Map.new(results)
            {:ok, result_map}

          {:ok, {:error, reason}, _} ->
            {:error, reason}

          {:timeout, _} ->
            {:error, :timeout}

          {:error, :noproc} ->
            {:error, :cluster_not_ready}

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

      duration = System.monotonic_time() - start_time

      :telemetry.execute(
        [:concord, :api, :touch_many],
        %{duration: duration},
        %{result: result, batch_size: length(operations)}
      )

      result
    end
  end

  # Private helpers

  # Helper function to process batch command results
  defp process_batch_command_result(command_result) do
    case command_result do
      {:ok, {:ok, results}, _} ->
        # Convert StateMachine results to map format
        success_map = Map.new(results, fn {key, :ok} -> {key, :ok} end)
        {:ok, success_map}

      {:ok, {:error, reason}, _} ->
        {:error, reason}

      {:timeout, _} ->
        {:error, :timeout}

      {:error, :noproc} ->
        {:error, :cluster_not_ready}

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

  defp command(cmd, timeout) do
    :ra.process_command(server_id(), cmd, timeout)
  end

  defp unwrap_command_result(cmd, timeout) do
    case command(cmd, timeout) do
      {:ok, :ok, _} -> :ok
      {:ok, {:error, reason}, _} -> {:error, reason}
      {:timeout, _} -> {:error, :timeout}
      {:error, :noproc} -> {:error, :cluster_not_ready}
      {:error, reason} -> {:error, reason}
    end
  end

  # Evaluate a condition function pre-consensus, then issue a CAS command.
  # Keeps anonymous functions out of the Raft log.
  defp evaluate_condition_then_cas(key, build_cmd, condition_fn, opts, timeout) do
    case get(key, Keyword.take(opts, [:timeout, :consistency])) do
      {:ok, current_value} ->
        if condition_fn.(current_value) do
          unwrap_command_result(build_cmd.(current_value), timeout)
        else
          {:error, :condition_failed}
        end

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

  defp query(query, timeout, consistency) do
    # Ra 3.0: leader_query and consistent_query require MFA tuples, not anonymous
    # functions (anonymous functions cannot be safely serialized across nodes).
    # Ra calls MFA as erlang:apply(M, F, A ++ [State]), so {M, :query, [q]}
    # maps to StateMachine.query(q, state) — matching query/2 directly.
    # local_query still accepts anonymous functions for eventual consistency.
    mfa = {StateMachine, :query, [query]}

    result =
      case consistency do
        :eventual ->
          # Use local_query for eventual consistency (fastest, may be stale)
          target_server = select_read_replica()

          :ra.local_query(
            target_server,
            fn state -> StateMachine.query(query, state) end,
            timeout
          )

        :leader ->
          # Use leader_query for balanced consistency
          :ra.leader_query(server_id(), mfa, timeout)

        :strong ->
          # Use consistent_query for linearizable reads (slowest)
          :ra.consistent_query(server_id(), mfa, timeout)

        _ ->
          # Default to leader query for unknown consistency levels
          :ra.leader_query(server_id(), mfa, timeout)
      end

    # Normalize the result format:
    # - local_query returns: {:ok, {{index, term}, result}, leader_id}
    # - leader_query returns: {:ok, {{index, term}, result}, leader_id} or {:ok, result, not_known}
    # - consistent_query returns: {:ok, result, leader_id}
    case result do
      {:ok, {{_index, _term}, query_result}, leader_id} ->
        # Standard format from local_query or leader_query
        {:ok, {{0, 0}, query_result}, leader_id}

      {:ok, query_result, leader_id} ->
        # Format from consistent_query or leader_query when leader unknown
        {:ok, {{0, 0}, query_result}, leader_id}

      other ->
        # Pass through errors and timeouts
        other
    end
  end

  defp server_id do
    {@cluster_name, node()}
  end

  defp default_consistency do
    Application.get_env(:concord, :default_read_consistency, :leader)
  end

  # Recursively convert data structures to JSON-friendly format
  # Converts tuples, references, PIDs, and other non-JSON types to strings
  defp make_json_friendly(data) when is_map(data) do
    Enum.map(data, fn {k, v} ->
      {make_json_friendly(k), make_json_friendly(v)}
    end)
    |> Map.new()
  end

  defp make_json_friendly(data) when is_list(data) do
    Enum.map(data, &make_json_friendly/1)
  end

  defp make_json_friendly(data) when is_tuple(data) do
    inspect(data)
  end

  defp make_json_friendly(data) when is_reference(data) or is_pid(data) or is_port(data) do
    inspect(data)
  end

  defp make_json_friendly(data), do: data

  defp select_read_replica do
    # Get cluster members for load balancing eventual consistency reads
    case :ra.members(server_id()) do
      {:ok, [_ | _] = members, _leader} ->
        # Randomly select a member for load balancing
        Enum.random(members)

      _ ->
        # Fallback to local server if we can't get members
        server_id()
    end
  end

  defp maybe_compress(value, opts) do
    case Keyword.get(opts, :compress) do
      true -> Compression.compress(value, force: true)
      false -> value
      nil -> Compression.compress(value)
    end
  end

  defp decompress_pairs(pairs) do
    Enum.map(pairs, fn {key, value} -> {key, Compression.decompress(value)} end)
  end

  defp validate_key(key) when is_binary(key) and byte_size(key) > 0 and byte_size(key) <= 1024 do
    :ok
  end

  defp validate_key(_), do: {:error, :invalid_key}

  defp validate_ttl_option(opts) do
    case Keyword.get(opts, :ttl) do
      nil -> :ok
      ttl -> TTL.validate_ttl(ttl)
    end
  end

  defp calculate_expires_at(nil), do: nil
  defp calculate_expires_at(:infinity), do: nil

  defp calculate_expires_at(ttl_seconds) when is_integer(ttl_seconds) and ttl_seconds > 0 do
    TTL.calculate_expiration(ttl_seconds)
  end

  # Batch operation validation helpers

  defp validate_batch_size(items) when is_list(items) do
    max_batch_size = Application.get_env(:concord, :max_batch_size, 500)

    if length(items) > max_batch_size do
      {:error, :batch_too_large}
    else
      :ok
    end
  end

  defp validate_batch_size(_), do: {:error, :invalid_batch_format}

  defp validate_put_operations(operations) when is_list(operations) do
    case Enum.find_value(operations, :ok, fn operation ->
           validate_put_operation(operation)
         end) do
      :ok -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  defp validate_put_operation({key, _value}) when is_binary(key) and byte_size(key) > 0 do
    validate_key(key)
  end

  defp validate_put_operation({key, _value, ttl}) when is_binary(key) and byte_size(key) > 0 do
    with :ok <- validate_key(key),
         :ok <- TTL.validate_ttl(ttl) do
      :ok
    end
  end

  defp validate_put_operation({key, _value, expires_at})
       when is_binary(key) and byte_size(key) > 0 do
    if expires_at == nil or is_integer(expires_at) do
      validate_key(key)
    else
      {:error, :invalid_expires_at}
    end
  end

  defp validate_put_operation(_), do: {:error, :invalid_operation_format}

  defp validate_keys(keys) when is_list(keys) do
    case Enum.find_value(keys, :ok, fn key ->
           validate_key(key)
         end) do
      :ok -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  defp validate_keys(_), do: {:error, :invalid_keys_format}

  defp validate_touch_operations(operations) when is_list(operations) do
    case Enum.find_value(operations, :ok, fn operation ->
           validate_touch_operation(operation)
         end) do
      :ok -> :ok
      {:error, reason} -> {:error, reason}
    end
  end

  defp validate_touch_operation({key, ttl_seconds})
       when is_binary(key) and byte_size(key) > 0 and is_integer(ttl_seconds) and ttl_seconds > 0 do
    :ok
  end

  defp validate_touch_operation(_), do: {:error, :invalid_touch_operation}

  defp validate_condition_opts(opts) do
    expected = Keyword.get(opts, :expected)
    condition_fn = Keyword.get(opts, :condition)

    cond do
      expected == nil and condition_fn == nil ->
        {:error, :missing_condition}

      expected != nil and condition_fn != nil ->
        {:error, :conflicting_conditions}

      is_function(condition_fn, 1) or expected != nil ->
        :ok

      true ->
        {:error, :invalid_condition}
    end
  end

  # ══════════════════════════════════════════════
  # v2 API convenience delegates
  # ══════════════════════════════════════════════

  @doc """
  Returns the current cluster revision.

  See `Concord.KV.revision/1`.
  """
  defdelegate revision(opts \\ []), to: Concord.KV

  @doc """
  Lists keys matching a prefix or range selector.

  See `Concord.KV.list/1`.
  """
  defdelegate list(opts), to: Concord.KV

  @doc """
  Commits a transaction atomically.

  See `Concord.Txn.commit/2`.
  """
  def txn(spec, opts \\ []) do
    Txn.commit(spec, opts)
  end
end