mongodb_driver

defmodule Mongo.ServerDescription do
  @moduledoc false

  alias Mongo.Version

  @retryable_wire_version Version.encode(:supports_op_msg)

  # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#serverdescription
  @type type :: :standalone | :mongos | :possible_primary | :rs_primary | :rs_secondary | :rs_arbiter | :rs_other | :rs_ghost | :unknown

  if Code.ensure_loaded?(:ezstd) do
    @type compressor_types :: :zlib | :zstd
    @support_compressors ["zlib", "zstd"]
  else
    @type compressor_types :: :zlib
    @support_compressors ["zlib"]
  end

  @type t :: %{
          address: String.t() | nil,
          error: String.t() | nil,
          round_trip_time: non_neg_integer | nil,
          last_write_date: DateTime.t(),
          op_time: BSON.ObjectId.t() | nil,
          type: type,
          min_wire_version: non_neg_integer,
          max_wire_version: non_neg_integer,
          me: String.t() | nil,
          hosts: [String.t()],
          passives: [String.t()],
          arbiters: [String.t()],
          tag_set: %{String.t() => String.t()},
          set_name: String.t() | nil,
          set_version: non_neg_integer | nil,
          election_id: BSON.ObjectId.t() | nil,
          primary: String.t() | nil,
          last_update_time: non_neg_integer,
          max_bson_object_size: non_neg_integer,
          max_message_size_bytes: non_neg_integer,
          max_write_batch_size: non_neg_integer,
          compression: [compressor_types],
          read_only: boolean(),
          logical_session_timeout: non_neg_integer,
          supports_retryable_writes: boolean(),
          replica?: boolean()
        }

  @empty %{
    address: "localhost:27017",
    error: nil,
    round_trip_time: nil,
    last_write_date: nil,
    op_time: nil,
    type: :unknown,
    min_wire_version: 0,
    max_wire_version: 0,
    me: nil,
    hosts: [],
    passives: [],
    arbiters: [],
    tag_set: %{},
    set_name: nil,
    set_version: nil,
    election_id: nil,
    primary: nil,
    last_update_time: 0,
    max_bson_object_size: 16_777_216,
    max_message_size_bytes: 48_000_000,
    max_write_batch_size: 100_000,
    compression: [],
    read_only: false,
    logical_session_timeout: 30,
    support_retryable_writes: false,
    replica?: false
  }

  def new() do
    @empty
  end

  def defaults(map \\ %{}) do
    Map.merge(@empty, map)
  end

  # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#parsing-an-ismaster-response
  def parse_hello_response(address, error) do
    Map.merge(@empty, %{address: address, error: error})
  end

  def parse_hello_response(hello_response) do
    server_type = determine_server_type(hello_response)
    max_wire_version = hello_response["maxWireVersion"] || 0

    supports_retryable_writes =
      server_type != :standalone &&
        max_wire_version >= @retryable_wire_version &&
        hello_response["logicalSessionTimeoutMinutes"] != nil

    %{
      type: server_type,
      last_write_date: get_in(hello_response, ["lastWrite", "lastWriteDate"]),
      op_time: get_in(hello_response, ["lastWrite", "opTime"]),
      min_wire_version: hello_response["minWireVersion"] || 0,
      max_wire_version: max_wire_version,
      me: hello_response["me"],
      hosts: (hello_response["hosts"] || []) |> Enum.map(&String.downcase/1),
      passives: (hello_response["passives"] || []) |> Enum.map(&String.downcase/1),
      arbiters: (hello_response["arbiters"] || []) |> Enum.map(&String.downcase/1),
      tag_set: hello_response["tags"] || %{},
      set_name: hello_response["setName"],
      set_version: hello_response["setVersion"],
      election_id: hello_response["electionId"],
      primary: hello_response["primary"],
      max_bson_object_size: hello_response["maxBsonObjectSize"] || 16_777_216,
      max_message_size_bytes: hello_response["maxMessageSizeBytes"] || 48_000_000,
      max_write_batch_size: hello_response["maxWriteBatchSize"] || 100_000,
      compression: map_compressors(hello_response["compression"]),
      read_only: hello_response["readOnly"] || false,
      logical_session_timeout: hello_response["logicalSessionTimeoutMinutes"] || 30,
      supports_retryable_writes: supports_retryable_writes,
      replica?: replica?(server_type)
    }
  end

  def parse_hello_response(address, rtt, last_update_time, hello_response) do
    server_type = determine_server_type(hello_response)
    max_wire_version = hello_response["maxWireVersion"] || 0

    %{
      address: address,
      round_trip_time: rtt,
      type: server_type,
      last_write_date: get_in(hello_response, ["lastWrite", "lastWriteDate"]),
      op_time: get_in(hello_response, ["lastWrite", "opTime"]),
      last_update_time: last_update_time,
      min_wire_version: hello_response["minWireVersion"] || 0,
      max_wire_version: max_wire_version,
      me: hello_response["me"],
      hosts: (hello_response["hosts"] || []) |> Enum.map(&String.downcase/1),
      passives: (hello_response["passives"] || []) |> Enum.map(&String.downcase/1),
      arbiters: (hello_response["arbiters"] || []) |> Enum.map(&String.downcase/1),
      tag_set: hello_response["tags"] || %{},
      set_name: hello_response["setName"],
      set_version: hello_response["setVersion"],
      election_id: hello_response["electionId"],
      primary: hello_response["primary"],
      max_bson_object_size: hello_response["maxBsonObjectSize"] || 16_777_216,
      max_message_size_bytes: hello_response["maxMessageSizeBytes"] || 48_000_000,
      max_write_batch_size: hello_response["maxWriteBatchSize"] || 100_000,
      compression: map_compressors(hello_response["compression"]),
      read_only: hello_response["readOnly"] || false,
      logical_session_timeout: hello_response["logicalSessionTimeoutMinutes"] || 30,
      supports_retryable_writes: server_type != :standalone && max_wire_version >= @retryable_wire_version && hello_response["logicalSessionTimeoutMinutes"] != nil,
      replica?: replica?(server_type)
    }
  end

  defp map_compressors(nil) do
    []
  end

  defp map_compressors(compressors) do
    compressors
    |> Enum.filter(fn compressor -> compressor in @support_compressors end)
    |> Enum.map(fn compressor -> String.to_existing_atom(compressor) end)
  end

  # see https://github.com/mongodb/specifications/blob/master/source/server-discovery-and-monitoring/server-discovery-and-monitoring.rst#type
  defp determine_server_type(%{"ok" => n}) when n != 1, do: :unknown
  defp determine_server_type(%{"msg" => "isdbgrid"}), do: :mongos
  defp determine_server_type(%{"isreplicaset" => true}), do: :rs_ghost

  defp determine_server_type(%{"setName" => set_name} = is_master_reply) when set_name != nil do
    case is_master_reply do
      %{"ismaster" => true} -> :rs_primary
      %{"isWritablePrimary" => true} -> :rs_primary
      %{"secondary" => true} -> :rs_secondary
      %{"arbiterOnly" => true} -> :rs_arbiter
      _ -> :rs_other
    end
  end

  defp determine_server_type(_), do: :standalone

  if Code.ensure_loaded?(:ezstd) do
    def support_compressors() do
      [:zstd, :zlib]
    end
  else
    def support_compressors() do
      [:zlib]
    end
  end

  defp replica?(server_type) do
    server_type in [:rs_primary, :rs_secondary, :rs_arbiter, :rs_other, :rs_ghost]
  end
end