electric

defmodule Electric.Config.Defaults do
  @moduledoc false

  # we want the default storage and kv implementations to honour the
  # `:storage_dir` configuration setting so we need to use runtime-evaluated
  # functions to get them. Since you can't embed anoymous functions these
  # functions are used instead.

  @doc false
  def storage(opts \\ []) do
    storage_dir = Keyword.get_lazy(opts, :storage_dir, fn -> storage_dir("shapes") end)
    {Electric.ShapeCache.PureFileStorage, storage_dir: storage_dir}
  end

  @doc false
  def persistent_kv(opts \\ []) do
    storage_dir = Keyword.get_lazy(opts, :storage_dir, fn -> storage_dir("state") end)
    {Electric.PersistentKV.Filesystem, :new!, root: storage_dir}
  end

  defp storage_dir(sub_dir) do
    Path.join(storage_dir(), sub_dir)
  end

  defp storage_dir do
    Electric.Config.get_env(:storage_dir)
  end

  def process_registry_partitions do
    System.schedulers_online()
  end
end

defmodule Electric.Config do
  require Logger

  @type instance_id :: String.t()

  @build_env Mix.env()

  @known_feature_flags ~w[allow_subqueries tagged_subqueries]
  @default_storage_dir "./persistent"

  @defaults [
    ## Database
    provided_database_id: "single_stack",
    db_pool_size: 20,
    replication_stream_id: "default",
    replication_slot_temporary?: false,
    replication_slot_temporary_random_name?: false,
    max_txn_size: 250 * 1024 * 1024,
    max_batch_size: 100,
    # Scaling down on idle is disabled by default
    replication_idle_timeout: 0,
    manual_table_publishing?: false,
    ## HTTP API
    # set enable_http_api: false to turn off the HTTP server totally
    enable_http_api: true,
    long_poll_timeout: 20_000,
    http_api_num_acceptors: nil,
    tcp_send_timeout: :timer.seconds(30),
    cache_max_age: 60,
    cache_stale_age: 60 * 5,
    chunk_bytes_threshold: Electric.ShapeCache.LogChunker.default_chunk_size_threshold(),
    allow_shape_deletion?: false,
    service_port: 3000,
    listen_on_ipv6?: false,
    stack_ready_timeout: 5_000,
    send_cache_headers?: true,
    max_shapes: nil,
    consumer_partitions: nil,
    # This value should be tuned for the hardware it's running on.
    max_concurrent_requests: %{initial: 300, existing: 10_000},
    ## Storage
    storage_dir: @default_storage_dir,
    storage: &Electric.Config.Defaults.storage/0,
    persistent_kv: &Electric.Config.Defaults.persistent_kv/0,
    cleanup_interval_ms: 10_000,
    ## Telemetry
    instance_id: nil,
    prometheus_port: nil,
    call_home_telemetry?: @build_env == :prod,
    telemetry_statsd_host: nil,
    telemetry_url: URI.new!("https://checkpoint.electric-sql.com"),
    otel_sampling_ratio: 0.01,
    metrics_sampling_ratio: 1,
    # When true, the OTel spans of empty/up-to-date shape-GET responses are tail-dropped
    # at export time to cut trace volume. See Electric.Telemetry.EmptyResponseSampler.
    drop_empty_response_spans?: false,
    ## Memory
    # After this duration of inactivity, consumer processes will hibernate
    # to allow garbage collection
    shape_hibernate_after: :timer.seconds(30),
    # If enabled, terminate (suspend) consumer processes after hibernating.
    # This frees memory more aggressively than hibernation alone.
    shape_enable_suspend?: false,
    # After hibernating, wait this duration before suspending (terminating).
    # Only applies when shape_enable_suspend? is true.
    shape_suspend_after: :timer.minutes(10),
    # Sets max_requests for Bandit handler processes:
    # https://hexdocs.pm/bandit/Bandit.html#t:http_1_options/0
    # "The maximum number of requests to serve in a single HTTP/{1,2}
    # connection before closing the connection"
    conn_max_requests: 50,
    # Sets fullsweep_after for Bandit handler processes.
    # Bandit reuses handler processes across requests (capped by conn_max_requests).
    # These processes can accumulate garbage on the old heap since BEAM's generational
    # GC only does fullsweep after a certain number of minor collections. This setting
    # forces a fullsweep after N minor collections to reclaim that memory.
    # See https://www.erlang.org/doc/apps/erts/erlang.html#spawn_opt/4
    handler_fullsweep_after: nil,
    ## Performance tweaks
    publication_alter_debounce_ms: 0,
    # allow for configuring per-process `Process.spawn_opt()`. In the form
    # %{process_id :: atom() => [Process.spawn_opt()]}
    # See `Process.flag/2`
    #
    # e.g. %{shape_log_collector: [min_heap_size: 1024 * 1024, min_bin_vheap_size: 1024 * 1024]}
    process_spawn_opts: %{},
    # Heap-size threshold (in BYTES) above which a consumer runs :erlang.garbage_collect()
    # after processing a transaction fragment.
    consumer_gc_heap_threshold: nil,
    ## Misc
    process_registry_partitions: &Electric.Config.Defaults.process_registry_partitions/0,
    feature_flags: if(Mix.env() == :test, do: @known_feature_flags, else: []),
    publication_refresh_period: 60_000,
    schema_reconciler_period: 60_000,
    snapshot_timeout_to_first_data: :timer.seconds(30),
    shape_db_exclusive_mode: false,
    shape_db_storage_dir: @default_storage_dir,
    # TODO: fix defaults to synchronous=NORMAL and shape_db_cache_size=2048
    shape_db_synchronous:
      Electric.ShapeCache.ShapeStatus.ShapeDb.Connection.default!(:synchronous),
    shape_db_cache_size: Electric.ShapeCache.ShapeStatus.ShapeDb.Connection.default!(:cache_size),
    exclude_spans: MapSet.new(),
    live_dashboard_port: nil,
    subquery_buffer_max_transactions: 1000
  ]

  @installation_id_key "electric_installation_id"

  def default(key) do
    case Keyword.fetch!(@defaults, key) do
      fun when is_function(fun, 0) -> fun.()
      value -> value
    end
  end

  @doc false
  @spec ensure_instance_id() :: instance_id()
  # the instance id needs to be consistent across calls, so we do need to have
  # a value in the config, even if it's not configured by the user.
  def ensure_instance_id do
    case Application.get_env(:electric, :instance_id) do
      nil ->
        instance_id = generate_instance_id()

        Logger.notice("Setting electric instance_id: #{instance_id}")
        Application.put_env(:electric, :instance_id, instance_id)

        instance_id

      id when is_binary(id) ->
        id
    end
  end

  defp generate_instance_id do
    Electric.Utils.uuid4()
  end

  # the installation id is persisted to disk to remain the same between restarts of the sync service
  @spec persist_installation_id(term, binary) :: instance_id()
  def persist_installation_id(persistent_kv, instance_id) when is_binary(instance_id) do
    case Electric.PersistentKV.get(persistent_kv, @installation_id_key) do
      {:ok, id} when is_binary(id) ->
        id

      {:error, :not_found} ->
        :ok = Electric.PersistentKV.set(persistent_kv, @installation_id_key, instance_id)
        instance_id
    end
  end

  @spec installation_id!(term) :: binary | no_return
  def installation_id!(kv) do
    case Electric.PersistentKV.get(kv, @installation_id_key) do
      {:ok, id} when is_binary(id) -> id
      {:error, :not_found} -> raise "Electric's installation_id not set"
    end
  end

  @doc """
  The minimum allowed time before Electric can close database connections due to the
  replication stream inactivity.

  This is to prevent churn where connection and replication supervisors would restart too frequently.

  The scale-to-zero feature of managed providers like Neon takes on the order of minutes before
  deciding that an idle database can be scaled down.
  """
  @spec min_replication_idle_timeout() :: pos_integer
  def min_replication_idle_timeout, do: 30_000

  def min_replication_idle_timeout_in_seconds, do: div(min_replication_idle_timeout(), 1000)

  @spec get_env(Application.key()) :: Application.value()
  def get_env(key) do
    # handle the case where the config value was set in runtime.exs but to
    # `nil` because of a missing env var. This allows us to just use `nil`
    # as the default config values in runtime.exs so avoiding hard-coding
    # defaults all over the place.
    case Application.get_env(:electric, key) do
      nil -> default(key)
      value -> value
    end
  end

  def get_env_lazy(key, fun) when is_function(fun, 0) do
    case Application.fetch_env(:electric, key) do
      {:ok, nil} -> fun.()
      {:ok, value} -> value
      :error -> fun.()
    end
  end

  @spec fetch_env!(Application.key()) :: Application.value()
  def fetch_env!(key) do
    Application.fetch_env!(:electric, key)
  end

  def persistent_kv do
    with {m, f, a} <- get_env(:persistent_kv) do
      apply(m, f, [a])
    end
  end

  @doc ~S"""
  Parse a PostgreSQL URI into a keyword list.

  ## Examples

      iex> parse_postgresql_uri("postgresql://postgres:password@example.com/app-db") |> deobfuscate()
      {:ok, [
        hostname: "example.com",
        port: 5432,
        database: "app-db",
        username: "postgres",
        password: "password",
      ]}

      iex> parse_postgresql_uri("postgresql://electric@192.168.111.33:81/__shadow")
      {:ok, [
        hostname: "192.168.111.33",
        port: 81,
        database: "__shadow",
        username: "electric"
      ]}

      iex> parse_postgresql_uri("postgresql://pg@[2001:db8::1234]:4321")
      {:ok, [
        hostname: "2001:db8::1234",
        port: 4321,
        database: "pg",
        username: "pg"
      ]}

      iex> parse_postgresql_uri("postgresql://user@localhost:5433/")
      {:ok, [
        hostname: "localhost",
        port: 5433,
        database: "user",
        username: "user"
      ]}

      iex> parse_postgresql_uri("postgresql://user%2Btesting%40gmail.com:weird%2Fpassword@localhost:5433/my%2Bdb%2Bname") |> deobfuscate()
      {:ok, [
        hostname: "localhost",
        port: 5433,
        database: "my+db+name",
        username: "user+testing@gmail.com",
        password: "weird/password"
      ]}

      iex> parse_postgresql_uri("postgres://super_user@localhost:7801/postgres?sslmode=disable")
      {:ok, [
        hostname: "localhost",
        port: 7801,
        database: "postgres",
        username: "super_user",
        sslmode: :disable
      ]}

      iex> parse_postgresql_uri("postgres://super_user@localhost:7801/postgres?sslmode=require")
      {:ok, [
        hostname: "localhost",
        port: 7801,
        database: "postgres",
        username: "super_user",
        sslmode: :require
      ]}

      iex> parse_postgresql_uri("postgres://super_user@localhost:7801/postgres?sslmode=yesplease")
      {:error, "invalid \"sslmode\" value: \"yesplease\""}

      iex> parse_postgresql_uri("postgrex://localhost")
      {:error, "invalid URL scheme: \"postgrex\""}

      iex> parse_postgresql_uri("postgresql://localhost")
      {:error, "invalid or missing username"}

      iex> parse_postgresql_uri("postgresql://:@localhost")
      {:error, "invalid or missing username"}

      iex> parse_postgresql_uri("postgresql://:password@localhost")
      {:error, "invalid or missing username"}

      iex> parse_postgresql_uri("postgresql://user:password")
      {:error, "invalid or missing username"}

      iex> parse_postgresql_uri("postgresql://user:password@")
      {:error, "missing host"}

      iex> parse_postgresql_uri("postgresql://user@localhost:5433/mydb?opts=-c%20synchronous_commit%3Doff&foo=bar")
      {:ok, [
        hostname: "localhost",
        port: 5433,
        database: "mydb",
        username: "user"
      ]}

      iex> parse_postgresql_uri("postgres://user:pass@localhost:5432/db?uselibpqcompat=true") |> deobfuscate()
      {:ok, [
        hostname: "localhost",
        port: 5432,
        database: "db",
        username: "user",
        password: "pass"
      ]}

      iex> parse_postgresql_uri("postgres://user:pass@localhost:5432/db?uselibpqcompat=true&sslmode=require") |> deobfuscate()
      {:ok, [
        hostname: "localhost",
        port: 5432,
        database: "db",
        username: "user",
        password: "pass",
        sslmode: :require
      ]}

      iex> parse_postgresql_uri("postgresql://electric@localhost/db?replication=database")
      {:error, "unsupported \"replication\" query option. Electric opens both a replication connection and regular connections to Postgres as needed"}

      iex> parse_postgresql_uri("postgresql://electric@localhost/db?replication=off")
      {:error, "unsupported \"replication\" query option. Electric opens both a replication connection and regular connections to Postgres as needed"}
  """
  @spec parse_postgresql_uri(binary) :: {:ok, keyword} | {:error, binary}
  def parse_postgresql_uri(uri_str) do
    %URI{scheme: scheme, host: host, port: port, path: path, userinfo: userinfo, query: query} =
      URI.parse(uri_str)

    with :ok <- validate_url_scheme(scheme),
         :ok <- validate_url_host(host),
         {:ok, {username, password}} <- parse_url_userinfo(userinfo),
         {:ok, options} <- parse_url_query(query) do
      conn_params =
        Enum.reject(
          [
            hostname: host,
            port: port || 5432,
            database: parse_database(path, username) |> URI.decode(),
            username: URI.decode(username),
            password: if(password, do: password |> URI.decode() |> Electric.Utils.wrap_in_fun())
          ] ++ options,
          fn {_key, val} -> is_nil(val) end
        )

      {:ok, conn_params}
    end
  end

  def parse_postgresql_uri!(uri_str) do
    case parse_postgresql_uri(uri_str) do
      {:ok, results} -> results
      {:error, message} -> raise Dotenvy.Error, message: message
    end
  end

  defp validate_url_scheme(scheme) when scheme in ["postgres", "postgresql"], do: :ok
  defp validate_url_scheme(scheme), do: {:error, "invalid URL scheme: #{inspect(scheme)}"}

  defp validate_url_host(str) do
    if is_binary(str) and String.trim(str) != "" do
      :ok
    else
      {:error, "missing host"}
    end
  end

  defp parse_url_userinfo(str) do
    with false <- is_nil(str),
         {:ok, {username, password}} <- split_userinfo(str),
         false <- String.trim(username) == "" do
      {:ok, {username, password}}
    else
      _ -> {:error, "invalid or missing username"}
    end
  end

  defp split_userinfo(str) do
    case String.split(str, ":") do
      [username] -> {:ok, {username, nil}}
      [username, password] -> {:ok, {username, password}}
      _ -> :error
    end
  end

  defp parse_url_query(nil), do: {:ok, []}

  defp parse_url_query(query_str) do
    params = URI.decode_query(query_str)

    with :ok <- validate_no_replication_param(params) do
      parse_sslmode(params)
    end
  end

  defp validate_no_replication_param(%{"replication" => _}) do
    {:error,
     "unsupported \"replication\" query option. Electric opens both a replication connection and regular connections to Postgres as needed"}
  end

  defp validate_no_replication_param(_), do: :ok

  defp parse_sslmode(%{"sslmode" => sslmode})
       when sslmode in ~w[disable allow prefer require] do
    {:ok, sslmode: String.to_existing_atom(sslmode)}
  end

  defp parse_sslmode(%{"sslmode" => sslmode})
       when sslmode in ~w[verify-ca verify-full] do
    {:error,
     "unsupported \"sslmode\" value #{inspect(sslmode)}. Use sslmode=require and set the ELECTRIC_DATABASE_CA_CERTIFICATE_FILE config to ensure Electric verifies database server identity"}
  end

  defp parse_sslmode(%{"sslmode" => sslmode}) do
    {:error, "invalid \"sslmode\" value: #{inspect(sslmode)}"}
  end

  defp parse_sslmode(_), do: {:ok, []}

  defp parse_database(nil, username), do: username
  defp parse_database("/", username), do: username
  defp parse_database("/" <> dbname, _username), do: dbname

  @log_levels ~w[emergency alert critical error warning warn notice info debug]
  @public_log_levels ~w[error warning info debug]

  @spec parse_log_level(binary) :: {:ok, Logger.level()} | {:error, binary}
  def parse_log_level(str) when str in @log_levels do
    {:ok, String.to_existing_atom(str)}
  end

  def parse_log_level(str) do
    {:error, "invalid log level: #{inspect(str)}. Must be one of #{inspect(@public_log_levels)}"}
  end

  def parse_log_level!(str) when str in @log_levels, do: String.to_existing_atom(str)

  def parse_log_level!(_str) do
    raise Dotenvy.Error, message: "Must be one of #{inspect(@public_log_levels)}"
  end

  @spec parse_telemetry_url(binary) :: {:ok, binary} | {:error, binary}
  def parse_telemetry_url(str) do
    case URI.new(str) do
      {:ok, %URI{scheme: scheme}} when scheme in ["http", "https"] -> {:ok, str}
      _ -> {:error, "invalid URL format: \"#{str}\""}
    end
  end

  def parse_telemetry_url!(str) do
    case parse_telemetry_url(str) do
      {:ok, url} -> url
      {:error, message} -> raise Dotenvy.Error, message: message
    end
  end

  @time_units ~w[ms msec s sec m min]

  @spec parse_human_readable_time(binary | nil) :: {:ok, pos_integer} | {:error, binary}

  def parse_human_readable_time(str) do
    with {num, suffix} <- Float.parse(str),
         true <- num > 0,
         suffix = String.trim(suffix),
         true <- suffix == "" or suffix in @time_units do
      {:ok, trunc(num * time_multiplier(suffix))}
    else
      _ -> {:error, "invalid time unit: #{inspect(str)}. Must be one of #{inspect(@time_units)}"}
    end
  end

  defp time_multiplier(""), do: 1
  defp time_multiplier(millisecond) when millisecond in ["ms", "msec"], do: 1
  defp time_multiplier(second) when second in ["s", "sec"], do: 1000
  defp time_multiplier(minute) when minute in ["m", "min"], do: 1000 * 60

  def parse_human_readable_time!(str) do
    case parse_human_readable_time(str) do
      {:ok, result} -> result
      {:error, message} -> raise Dotenvy.Error, message: message
    end
  end

  @doc """
  Parse human-readable memory/storage size string into bytes.

  ## Examples

    iex> parse_human_readable_size("1GiB")
    {:ok, #{1024 * 1024 * 1024}}

    iex> parse_human_readable_size("2.23GB")
    {:ok, 2_230_000_000}

    iex> parse_human_readable_size("256MiB")
    {:ok, #{256 * 1024 * 1024}}

    iex> parse_human_readable_size("377MB")
    {:ok, 377_000_000}

    iex> parse_human_readable_size("430KiB")
    {:ok, #{430 * 1024}}

    iex> parse_human_readable_size("142888KB")
    {:ok, 142_888_000}

    iex> parse_human_readable_size("123456789")
    {:ok, 123_456_789}

    iex> parse_human_readable_size("")
    {:error, ~S'invalid size unit: "". Must be one of ["KB", "KiB", "MB", "MiB", "GB", "GiB"]'}

    iex> parse_human_readable_size("foo")
    {:error, ~S'invalid size unit: "foo". Must be one of ["KB", "KiB", "MB", "MiB", "GB", "GiB"]'}
  """
  @spec parse_human_readable_size(binary) :: {:ok, pos_integer} | {:error, binary}

  @size_units ~w[KB KiB MB MiB GB GiB]

  def parse_human_readable_size(str) do
    with {num, suffix} <- Float.parse(str),
         true <- num > 0,
         suffix = String.trim(suffix),
         true <- suffix == "" or suffix in @size_units do
      {:ok, trunc(num * size_multiplier(suffix))}
    else
      _ -> {:error, "invalid size unit: #{inspect(str)}. Must be one of #{inspect(@size_units)}"}
    end
  end

  defp size_multiplier(""), do: 1
  defp size_multiplier("KB"), do: 1_000
  defp size_multiplier("KiB"), do: 1024
  defp size_multiplier("MB"), do: 1_000_000
  defp size_multiplier("MiB"), do: 1024 * 1024
  defp size_multiplier("GB"), do: 1_000_000_000
  defp size_multiplier("GiB"), do: 1024 * 1024 * 1024

  def parse_human_readable_size!(str) do
    case parse_human_readable_size(str) do
      {:ok, result} -> result
      {:error, message} -> raise Dotenvy.Error, message: message
    end
  end

  @doc """
  Parse a top process limit string into a tagged tuple.

  ## Examples

    iex> parse_top_process_limit("count:5")
    {:ok, {:count, 5}}

    iex> parse_top_process_limit("mem_percent:80")
    {:ok, {:mem_percent, 80}}

    iex> parse_top_process_limit("mem_percent:0")
    {:error, "mem_percent value must be between 1 and 100, got: 0"}

    iex> parse_top_process_limit("count:0")
    {:error, "count value must be a positive integer, got: 0"}

    iex> parse_top_process_limit("foo")
    {:error, ~S'invalid top process limit: "foo". Expected format: count:<N> or mem_percent:<N>'}
  """
  @spec parse_top_process_limit(binary) ::
          {:ok, {:count, pos_integer()} | {:mem_percent, 1..100}} | {:error, binary}
  def parse_top_process_limit(str) do
    case String.split(str, ":", parts: 2) do
      ["count", n] ->
        parse_process_count(n)

      ["mem_percent", n] ->
        parse_mem_percent(n)

      _ ->
        {:error,
         "invalid top process limit: #{inspect(str)}. Expected format: count:<N> or mem_percent:<N>"}
    end
  end

  def parse_top_process_limit!(str) do
    case parse_top_process_limit(str) do
      {:ok, result} -> result
      {:error, message} -> raise Dotenvy.Error, message: message
    end
  end

  @doc """
  Parse the deprecated ELECTRIC_TELEMETRY_TOP_PROCESS_COUNT value (a plain integer)
  into a `{:count, N}` tuple.

  ## Examples

    iex> parse_legacy_top_process_count("10")
    {:ok, {:count, 10}}

    iex> parse_legacy_top_process_count("0")
    {:error, "count value must be a positive integer, got: 0"}
  """
  @spec parse_legacy_top_process_count(binary) ::
          {:ok, {:count, pos_integer()}} | {:error, binary}
  def parse_legacy_top_process_count(str), do: parse_process_count(str)

  def parse_legacy_top_process_count!(str) do
    case parse_legacy_top_process_count(str) do
      {:ok, result} -> result
      {:error, message} -> raise Dotenvy.Error, message: message
    end
  end

  defp parse_process_count(str) do
    case Integer.parse(str) do
      {val, ""} when val > 0 -> {:ok, {:count, val}}
      _ -> {:error, "count value must be a positive integer, got: #{str}"}
    end
  end

  defp parse_mem_percent(str) do
    case Integer.parse(str) do
      {val, ""} when val >= 1 and val <= 100 -> {:ok, {:mem_percent, val}}
      _ -> {:error, "mem_percent value must be between 1 and 100, got: #{str}"}
    end
  end

  def validate_security_config!(secret, insecure) do
    cond do
      insecure && secret != nil ->
        raise "You cannot set both ELECTRIC_SECRET and ELECTRIC_INSECURE=true"

      !insecure && secret == nil ->
        raise "You must set ELECTRIC_SECRET unless ELECTRIC_INSECURE=true. Setting ELECTRIC_INSECURE=true risks exposing your database, only use insecure mode in development or you've otherwise secured the Electric API"

      true ->
        if insecure do
          Logger.warning(
            "Electric is running in insecure mode - this risks exposing your database - only use insecure mode in development or if you've otherwise secured the Electric API."
          )
        end

        :ok
    end
  end

  @valid_spawn_opts ~w[min_bin_vheap_size min_heap_size priority fullsweep_after message_queue_data]

  @doc """
  Parse `spawn_opts` from environment variable to keyword list suitable for passing to `GenServer.start_link/2`

  ## Examples

      iex> parse_spawn_opts!(~S({"shape_log_collector":{"min_heap_size":234,"min_bin_vheap_size":123,"message_queue_data":"on_heap","priority":"high","fullsweep_after":104}}))
      %{shape_log_collector: [fullsweep_after: 104, message_queue_data: :on_heap, min_bin_vheap_size: 123, min_heap_size: 234, priority: :high]}

      iex> parse_spawn_opts!(~S({"shape_log_collector":{"monkey":123,"message_queue_data":"on_fire","min_bin_vheap_size":-1,"priority":"high"}}))
      %{shape_log_collector: [priority: :high]}

      iex> parse_spawn_opts!("")
      %{}

      iex> parse_spawn_opts!("{}")
      %{}
  """
  def parse_spawn_opts!("") do
    %{}
  end

  def parse_spawn_opts!(str) do
    str
    |> Jason.decode!()
    |> then(fn
      opts when is_map(opts) ->
        for {process_name, process_opts} <- opts, is_map(process_opts), into: %{} do
          opts =
            for {opt_key, opt_val} <- process_opts,
                opt_key in @valid_spawn_opts,
                key = String.to_atom(opt_key),
                val = validate_spawn_opt(key, opt_val) do
              {key, val}
            end

          {String.to_atom(process_name), opts}
        end

      _invalid ->
        raise ArgumentError, message: "Invalid spawn opts: #{inspect(str)}"
    end)
    |> tap(fn process_spawn_opts ->
      if map_size(process_spawn_opts) > 0 do
        Logger.notice("Process spawn opts: #{inspect(process_spawn_opts)}")
      end
    end)
  end

  defp validate_spawn_opt(:min_bin_vheap_size, val) when is_integer(val) and val >= 0, do: val
  defp validate_spawn_opt(:min_heap_size, val) when is_integer(val) and val >= 0, do: val

  defp validate_spawn_opt(:priority, val) when val in ["low", "normal", "high"],
    do: String.to_atom(val)

  defp validate_spawn_opt(:fullsweep_after, val) when is_integer(val) and val >= 0, do: val

  defp validate_spawn_opt(:message_queue_data, val) when val in ["off_heap", "on_heap"],
    do: String.to_atom(val)

  defp validate_spawn_opt(_, _), do: nil

  @doc false
  # helper function for use in doc tests
  def deobfuscate({:ok, connection_opts}),
    do: {:ok, Electric.Utils.deobfuscate_password(connection_opts)}

  def deobfuscate(other), do: other

  @doc ~S"""
  Parse a comma-separated string into a MapSet of trimmed values.

  ## Examples

      iex> parse_comma_separated_set!("foo,bar,baz")
      MapSet.new(["bar", "baz", "foo"])

      iex> parse_comma_separated_set!(" foo , bar , baz ")
      MapSet.new(["bar", "baz", "foo"])

      iex> parse_comma_separated_set!("single")
      MapSet.new(["single"])

      iex> parse_comma_separated_set!(",,,")
      MapSet.new()

      iex> parse_comma_separated_set!("")
      MapSet.new()
  """
  def parse_comma_separated_set!(str) do
    str |> String.split(",", trim: true) |> Enum.map(&String.trim/1) |> MapSet.new()
  end

  def parse_feature_flags(str) do
    str
    |> String.split(",")
    |> Enum.map(&String.trim/1)
    |> Enum.reject(&(&1 == ""))
    |> Enum.split_with(&(&1 in @known_feature_flags))
    |> case do
      {known, []} ->
        known

      {known, unknown} ->
        # Log an error but don't raise so that deployments can proceeed without
        # removal of old flags and new flags can be added before deployment
        Logger.error(
          "Unknown feature flags specified: #{inspect(unknown)}. Known feature flags: #{inspect(@known_feature_flags)}"
        )

        known
    end
  end
end