nebulex_distributed

defmodule Nebulex.Adapters.Partitioned.Options do
  @moduledoc false

  # Compilation time options
  compile_opts = [
    primary_storage_adapter: [
      type: :atom,
      required: false,
      default: Nebulex.Adapters.Local,
      doc: """
      The adapter module used for the primary (local) storage on each cluster
      node. The partitioned adapter is a distributed wrapper that routes
      requests to the appropriate node based on consistent hashing. The actual
      data storage is handled by the primary storage adapter on each node.
      This option allows you to choose which adapter to use for this local
      storage. The configuration for the primary adapter is specified via the
      `:primary` start option.
      """
    ]
  ]

  # Start options
  start_opts = [
    primary: [
      type: :keyword_list,
      required: false,
      default: [],
      doc: """
      Configuration options passed to the primary storage adapter specified via
      `:primary_storage_adapter`. The available options depend on which adapter
      you choose. Refer to the documentation of your chosen primary storage
      adapter for the complete list of supported options.
      """
    ],
    hash_ring: [
      type: :keyword_list,
      required: false,
      default: [],
      doc: """
      Configuration options for the consistent hash ring used to distribute keys
      across cluster nodes. The hash ring maps each key to a node using virtual
      nodes (vnodes), which enables:

        * Minimal key redistribution when nodes join or leave.
        * Even distribution of keys across nodes.
        * Efficient node lookup for cache operations.

      See [`ExHashRing.Ring.start_link/2`][ex_hash_ring] for the complete list
      of supported options.

      [ex_hash_ring]: https://hexdocs.pm/ex_hash_ring/ExHashRing.Ring.html#start_link/1
      """
    ],
    rejoin_interval: [
      type: :timeout,
      required: false,
      default: :timer.seconds(30),
      doc: """
      The interval in **milliseconds** at which the `RingMonitor` periodically
      rejoins the `:pg` group to force ring synchronization across all cluster
      nodes.

      **Purpose:** This mechanism helps handle race conditions during concurrent
      node startups by ensuring all nodes eventually have a consistent view of
      the hash ring. Even if some join events are missed during initial cluster
      formation, each rejoin triggers new notifications that force ring updates.

      **Trade-offs:**

        * **Shorter intervals** (e.g., 10 seconds):
          - Faster consistency convergence.
          - More overhead from frequent rejoin events and notifications.
          - Better for highly dynamic clusters with frequent node changes.

        * **Longer intervals** (e.g., 60 seconds):
          - Lower overhead and reduced network traffic.
          - Slower eventual consistency.
          - Fine for stable clusters that don't change frequently.

      **Default (30 seconds):** Works well for most use cases, balancing
      consistency and overhead.
      """
    ],
    node_filter: [
      type: {:fun, 1},
      required: false,
      doc: """
      An optional 1-arity function that filters which cluster nodes are
      added to the hash ring. The function receives a node name
      (`t:node/0`) and must return `true` to include the node in the
      ring, or `false` to exclude it. Only nodes present in the ring
      will be selected to cache data.

      Excluded nodes are still part of the cache cluster (`:pg` group),
      so the cache remains fully usable from them — reads, writes, and
      all other operations work transparently, routing to ring nodes
      as usual.

      By default, all nodes that join the cluster are added to the
      hash ring.

      > #### Function Captures {: .info}
      >
      > Due to how anonymous functions are implemented in the Erlang VM,
      > it is best to use function captures (`&Mod.fun/1`) as node filters.
      > In other words, avoid using literal anonymous functions
      > (`fn ... -> ... end`) or local function captures (`&my_filter/1`)
      > as they cannot be serialized across distributed nodes.

      See the ["Node Filter"](`Nebulex.Adapters.Partitioned#module-node-filter`)
      section for more information and examples.
      """
    ]
  ]

  # Common runtime options
  common_runtime_opts = [
    timeout: [
      type: :timeout,
      required: false,
      default: 5000,
      doc: """
      The time in **milliseconds** to wait for a cache command to finish.

      This timeout applies to RPC calls made to remote nodes during partitioned
      cache operations. Since the partitioned adapter routes requests across
      cluster nodes, network latency and node load affect execution time.

      Set to `:infinity` to wait indefinitely. If a timeout occurs, the
      operation fails with an error. Note that the underlying cache operation
      may still complete on the remote node asynchronously.
      """
    ]
  ]

  # Stream options
  stream_opts = [
    on_error: [
      type: {:in, [:nothing, :raise]},
      type_doc: "`:raise` | `:nothing`",
      required: false,
      default: :raise,
      doc: """
      Controls error handling during stream evaluation across cluster nodes.

      When streaming entries from a partitioned cache, the adapter evaluates the
      stream on each cluster node. Since this involves RPC calls to remote
      nodes, failures can occur due to:

        * Network issues or RPC timeouts.
        * Errors on the remote node.
        * Temporary node unavailability.

      **Options:**

        * `:raise` (default) - Raise an exception immediately when an error
          occurs on any node. The stream evaluation stops, and no further nodes
          are queried.

        * `:nothing` - Skip errors silently and continue. Returns only
          successful results from nodes that responded without errors.
          Useful for resilience in environments where temporary node
          failures are acceptable.

      """
    ]
  ]

  # Nebulex common options
  @nbx_start_opts Nebulex.Cache.Options.__compile_opts__() ++ Nebulex.Cache.Options.__start_opts__()

  # Compilation time option schema
  @compile_opts_schema NimbleOptions.new!(compile_opts)

  # Start options schema
  @start_opts_schema NimbleOptions.new!(start_opts)

  # Common runtime options schema
  @common_runtime_opts_schema NimbleOptions.new!(common_runtime_opts)

  # Stream options schema
  @stream_opts_schema NimbleOptions.new!(stream_opts ++ common_runtime_opts)

  # Stream options docs schema
  @stream_opts_docs_schema NimbleOptions.new!(stream_opts)

  ## Docs API

  # coveralls-ignore-start

  @spec compile_options_docs() :: binary()
  def compile_options_docs do
    NimbleOptions.docs(@compile_opts_schema)
  end

  @spec start_options_docs() :: binary()
  def start_options_docs do
    NimbleOptions.docs(@start_opts_schema)
  end

  @spec common_runtime_options_docs() :: binary()
  def common_runtime_options_docs do
    NimbleOptions.docs(@common_runtime_opts_schema)
  end

  @spec stream_options_docs() :: binary()
  def stream_options_docs do
    NimbleOptions.docs(@stream_opts_docs_schema)
  end

  # coveralls-ignore-stop

  ## Validation API

  @spec validate_compile_opts!(keyword()) :: keyword()
  def validate_compile_opts!(opts) do
    NimbleOptions.validate!(opts, @compile_opts_schema)
  end

  @spec validate_start_opts!(keyword()) :: keyword()
  def validate_start_opts!(opts) do
    adapter_opts =
      opts
      |> Keyword.drop(@nbx_start_opts)
      |> NimbleOptions.validate!(@start_opts_schema)

    Keyword.merge(opts, adapter_opts)
  end

  @spec validate_common_runtime_opts!(keyword()) :: keyword()
  def validate_common_runtime_opts!(opts) do
    adapter_opts =
      opts
      |> Keyword.take([:timeout])
      |> NimbleOptions.validate!(@common_runtime_opts_schema)

    Keyword.merge(opts, adapter_opts)
  end

  @spec validate_stream_opts!(keyword()) :: keyword()
  def validate_stream_opts!(opts) do
    adapter_opts =
      opts
      |> Keyword.take([:timeout, :on_error])
      |> NimbleOptions.validate!(@stream_opts_schema)

    Keyword.merge(opts, adapter_opts)
  end
end