defmodule Mongo.MongoDBConnection do
  @moduledoc """
  Implementation of the DBConnection behaviour module.
  """

  require Logger

  use DBConnection
  use Mongo.Messages

  alias Mongo.Events
  alias Mongo.Events.CommandStartedEvent
  alias Mongo.MongoDBConnection.Utils
  alias Mongo.Events.MoreToComeEvent
  alias Mongo.StableVersion

  @timeout 15_000
  @find_one_flags ~w(slave_ok exhaust partial)a
  @write_concern ~w(w j wtimeout)a
  @insecure_cmds [:authenticate, :saslStart, :saslContinue, :getnonce, :createUser, :updateUser, :copydbgetnonce, :copydbsaslstart, :copydb, :isMaster, :ismaster, :hello]

  @impl true
  def connect(opts) do
    # Trap exits so that DBConnection calls `disconnect` on unexpected shutdowns
    Process.flag(:trap_exit, true)

    {write_concern, opts} = Keyword.split(opts, @write_concern)
    write_concern = Keyword.put_new(write_concern, :w, 1)

    state = %{
      connection: nil,
      request_id: 0,
      timeout: opts[:timeout] || @timeout,
      connect_timeout: opts[:connect_timeout] || @timeout,
      database: Keyword.fetch!(opts, :database),
      write_concern: Map.new(write_concern),
      wire_version: 0,
      auth_mechanism: opts[:auth_mechanism] || nil,
      connection_type: Keyword.fetch!(opts, :connection_type),
      server_pid: Keyword.get(opts, :server_pid),
      topology_pid: Keyword.fetch!(opts, :topology_pid),
      stable_api: Keyword.get(opts, :stable_api),
      use_op_msg: Keyword.get(opts, :stable_api) != nil,
      hello_ok: Keyword.get(opts, :stable_api) != nil,
      ssl: opts[:ssl] || opts[:tls] || false
    }

    connect(opts, state)
  end

  @impl true
  ## the stream monitor disconnects, we change the mode of the parent monitor
  def disconnect(_error, %{connection: {mod, socket}, connection_type: :stream_monitor, parent_pid: parent_pid}) do
    ## Logger.debug("MongoDB-Connection: disconnected stream monitor: #{inspect(error)}")
    GenServer.cast(parent_pid, :stop_streaming_mode)
    mod.close(socket)
    :ok
  end

  def disconnect(_error, %{connection: {mod, socket}, connection_type: :monitor, topology_pid: topology_pid, host: host, server_pid: server_pid}) do
    ## Logger.debug("MongoDB-Connection: disconnected: #{inspect(error)}, #{inspect(server_pid)}, #{inspect(host)}, cast disconnect :monitor")
    GenServer.cast(server_pid, :stop_streaming_mode)
    GenServer.cast(topology_pid, {:disconnect, :monitor, host, server_pid})
    mod.close(socket)
    :ok
  end

  def disconnect(_error, %{connection: {mod, socket}}) do
    ## Logger.debug("MongoDB-Connection: disconnected: #{inspect error}, #{inspect type}, #{inspect host} #{inspect server_pid}")
    mod.close(socket)
    :ok
  end

  @impl true
  def checkout(state), do: {:ok, state}
  @impl true
  def handle_begin(_opts, state), do: {:ok, nil, state}
  @impl true
  def handle_close(_query, _opts, state), do: {:ok, nil, state}
  @impl true
  def handle_commit(_opts, state), do: {:ok, nil, state}
  @impl true
  def handle_deallocate(_query, _cursor, _opts, state), do: {:ok, nil, state}
  @impl true
  def handle_declare(query, _params, _opts, state), do: {:ok, query, nil, state}
  @impl true
  def handle_fetch(_query, _cursor, _opts, state), do: {:halt, nil, state}
  @impl true
  def handle_prepare(query, _opts, state), do: {:ok, query, state}
  @impl true
  def handle_rollback(_opts, state), do: {:ok, nil, state}
  @impl true
  def handle_status(_opts, state), do: {:idle, state}

  @impl true
  def ping(%{connection_type: :client} = state) do
    cmd = [ping: 1]

    case Utils.command(-1, cmd, state) do
      {:ok, _flags, %{"ok" => ok}} when ok == 1 ->
        {:ok, state}

      {:ok, _flags, %{"ok" => ok, "errmsg" => msg, "code" => code}} when ok == 0 ->
        err = Mongo.Error.exception(message: msg, code: code)
        {:disconnect, err, state}

      {:disconnect, _, _} = error ->
        error
    end
  end

  @impl true
  def ping(state) do
    {:ok, state}
  end

  @impl true
  def handle_execute(%Mongo.Query{action: action} = query, _params, opts, original_state) do
    tmp_state = %{original_state | database: Keyword.get(opts, :database, original_state.database)}

    with {:ok, reply, tmp_state} <- send_command(action, opts, tmp_state) do
      {:ok, query, reply, Map.put(tmp_state, :database, original_state.database)}
    end
  end

  defp connect(opts, state) do
    result =
      with {:ok, state} <- tcp_connect(opts, state),
           {:ok, state} <- maybe_ssl(opts, state),
           {:ok, state} <- hand_shake(opts, state) do
        maybe_auth(opts, state)
      end

    case result do
      {:ok, state} ->
        {:ok, state}

      {:disconnect, reason, state} ->
        reason =
          case reason do
            {:tcp_recv, reason} -> Mongo.Error.exception(tag: :tcp, action: "recv", reason: reason, host: state.host)
            {:tcp_send, reason} -> Mongo.Error.exception(tag: :tcp, action: "send", reason: reason, host: state.host)
            %Mongo.Error{} = reason -> reason
          end

        {mod, socket} = state.connection
        mod.close(socket)
        {:error, reason}

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

  defp maybe_auth(opts, state) do
    case opts[:skip_auth] do
      true -> {:ok, state}
      _ -> Mongo.Auth.run(opts, state)
    end
  end

  defp maybe_ssl(opts, %{ssl: true} = state), do: ssl(opts, state)
  defp maybe_ssl(_opts, state), do: {:ok, state}

  defp ssl(opts, %{connection: {:gen_tcp, socket}} = state) do
    host = (opts[:hostname] || "localhost") |> to_charlist
    ssl_opts = Keyword.put_new(opts[:ssl_opts] || [], :server_name_indication, host)

    case :ssl.connect(socket, ssl_opts, state.connect_timeout) do
      {:ok, ssl_sock} ->
        {:ok, %{state | connection: {:ssl, ssl_sock}}}

      {:error, reason} ->
        :gen_tcp.close(socket)
        {:error, Mongo.Error.exception(tag: :ssl, action: "connect", reason: reason, host: state.host)}
    end
  end

  defp tcp_connect(opts, s) do
    {host, port} = Utils.hostname_port(opts)
    sock_opts = [:binary, active: false, packet: :raw, nodelay: true] ++ (opts[:socket_options] || [])

    s =
      case host do
        {:local, socket} -> Map.put(s, :host, socket)
        hostname -> Map.put(s, :host, "#{hostname}:#{port}")
      end

    case :gen_tcp.connect(host, port, sock_opts, s.connect_timeout) do
      {:ok, socket} ->
        # A suitable :buffer is only set if :recbuf is included in
        # :socket_options.
        {:ok, [sndbuf: sndbuf, recbuf: recbuf, buffer: buffer]} = :inet.getopts(socket, [:sndbuf, :recbuf, :buffer])

        buffer =
          buffer
          |> max(sndbuf)
          |> max(recbuf)

        :ok = :inet.setopts(socket, buffer: buffer)

        {:ok, %{s | connection: {:gen_tcp, socket}}}

      {:error, reason} ->
        {:error, Mongo.Error.exception(tag: :tcp, action: "connect", reason: reason, host: s.host)}
    end
  end

  defp hand_shake(opts, state) do
    cmd = handshake_command(state, client(opts[:appname] || "elixir-driver"))

    case Utils.command(-1, cmd, state) do
      {:ok, _flags, %{"ok" => ok, "maxWireVersion" => version} = response} when ok == 1 ->
        {:ok, %{state | wire_version: version, use_op_msg: version >= 6, hello_ok: Map.get(response, "helloOk", false)}}

      {:ok, _flags, %{"ok" => ok}} when ok == 1 ->
        {:ok, %{state | wire_version: 0}}

      {:ok, _flags, %{"ok" => ok, "errmsg" => msg, "code" => code}} when ok == 0 ->
        err = Mongo.Error.exception(message: msg, code: code)
        {:disconnect, err, state}

      {:disconnect, _, _} = error ->
        error
    end
  end

  defp client(app_name) do
    driver_version =
      case :application.get_key(:mongodb_driver, :vsn) do
        {:ok, version} -> to_string(version)
        _ -> "??"
      end

    {architecture, name} = get_architecture()

    version =
      case :os.version() do
        {one, two, tree} -> to_string(one) <> "." <> to_string(two) <> "." <> to_string(tree)
        s -> s
      end

    platform = "Elixir (" <> System.version() <> "), Erlang/OTP (" <> to_string(:erlang.system_info(:otp_release)) <> "), ERTS (" <> to_string(:erlang.system_info(:version)) <> ")"

    type = elem(:os.type(), 1)

    %{
      client: %{
        application: %{name: app_name}
      },
      driver: %{
        name: "mongodb_driver",
        version: driver_version
      },
      os: %{
        type: type,
        name: pretty_name(name),
        architecture: architecture,
        version: version
      },
      platform: platform
    }
  end

  defp get_architecture() do
    case String.split(to_string(:erlang.system_info(:system_architecture)), "-") do
      [architecture, name | _rest] -> {architecture, name}
      ["win32"] -> {"win32", "Windows"}
      [one] -> {"??", one}
      [] -> {"??", "??"}
    end
  end

  defp pretty_name("apple"), do: "Mac OS X"
  defp pretty_name(name), do: name

  defp provide_cmd_data([{command_name, _} | _] = cmd) do
    case Enum.member?(@insecure_cmds, command_name) do
      true -> {command_name, %{}}
      false -> {command_name, cmd}
    end
  end

  ##
  # Executes a hello or the legacy hello command
  ##
  defp send_command({:exec_hello, cmd}, opts, %{use_op_msg: true} = state) do
    db = opts[:database] || state.database
    timeout = Keyword.get(opts, :timeout, state.timeout)
    flags = Keyword.get(opts, :flags, 0x0)

    cmd = hello_command(cmd, state) ++ ["$db": db]

    event = %CommandStartedEvent{
      command: :hello,
      command_name: :hello,
      database_name: db,
      request_id: state.request_id,
      operation_id: opts[:operation_id],
      connection_id: self()
    }

    Events.notify(event, :commands)

    op = op_msg(flags: flags, sections: [section(payload_type: 0, payload: payload(doc: cmd))])

    case :timer.tc(fn -> Utils.post_request(op, state.request_id, %{state | timeout: timeout}) end) do
      {duration, {:ok, flags, doc}} ->
        state = %{state | request_id: state.request_id + 1}
        {:ok, {doc, event, flags, duration}, state}

      {_duration, error} ->
        error
    end
  end

  ##
  # Executes a more to come command
  ##
  defp send_command(:more_to_come, opts, %{use_op_msg: true} = state) do
    event = %MoreToComeEvent{command: :more_to_come, command_name: opts[:command_name] || :more_to_come}

    timeout = Keyword.get(opts, :timeout, state.timeout)

    Events.notify(event, :commands)

    case :timer.tc(fn -> Utils.get_response(state.request_id, %{state | timeout: timeout}) end) do
      {duration, {:ok, flags, doc}} ->
        {:ok, {doc, event, flags, duration}, state}

      {_duration, error} ->
        error
    end
  end

  defp send_command({:command, cmd}, opts, %{use_op_msg: true} = state) do
    {command_name, data} = provide_cmd_data(cmd)
    db = opts[:database] || state.database
    cmd = cmd ++ ["$db": db]
    flags = Keyword.get(opts, :flags, 0x0)

    # MongoDB 3.6 only allows certain command arguments to be provided this way. These are:
    op =
      case pulling_out?(cmd, :documents) || pulling_out?(cmd, :updates) || pulling_out?(cmd, :deletes) do
        nil -> op_msg(flags: flags, sections: [section(payload_type: 0, payload: payload(doc: cmd))])
        key -> pulling_out(cmd, flags, key)
      end

    # overwrite temporary timeout by timeout option
    timeout = Keyword.get(opts, :timeout, state.timeout)

    event = %CommandStartedEvent{
      command: data,
      command_name: opts[:command_name] || command_name,
      database_name: db,
      request_id: state.request_id,
      operation_id: opts[:operation_id],
      connection_id: self()
    }

    Events.notify(event, :commands)

    metadata = %{
      type: :mongodb_driver,
      command: data,
      command_name: opts[:command_name] || command_name,
      database_name: db,
      request_id: state.request_id,
      operation_id: opts[:operation_id],
      connection_id: self(),
      options: Keyword.get(opts, :telemetry_options, [])
    }

    :telemetry.execute([:mongodb_driver, :start], %{}, metadata)

    case :timer.tc(fn -> Utils.post_request(op, state.request_id, %{state | timeout: timeout}) end) do
      {duration, {:ok, flags, doc}} ->
        state = %{state | request_id: state.request_id + 1}
        {:ok, {doc, event, flags, duration}, state}

      {_duration, error} ->
        error
    end
  end

  defp send_command({:command, cmd}, opts, state) do
    [{command_name, _} | _] = cmd

    event = %CommandStartedEvent{
      command: cmd,
      command_name: opts[:command_name] || command_name,
      database_name: opts[:database] || state.database,
      request_id: state.request_id,
      operation_id: opts[:operation_id],
      connection_id: self()
    }

    flags = Keyword.take(opts, @find_one_flags)
    op = op_query(coll: Utils.namespace("$cmd", state, opts[:database]), query: cmd, select: "", num_skip: 0, num_return: 1, flags: flags(flags))
    timeout = Keyword.get(opts, :timeout, state.timeout)

    case :timer.tc(fn -> Utils.post_request(op, state.request_id, %{state | timeout: timeout}) end) do
      {duration, {:ok, flags, doc}} ->
        state = %{state | request_id: state.request_id + 1}
        {:ok, {doc, event, flags, duration}, state}

      {_duration, error} ->
        error
    end
  end

  defp send_command(:error, _opts, state) do
    exception = Mongo.Error.exception("Test-case")
    {:disconnect, exception, state}
  end

  defp pulling_out?(cmd, key) do
    case Keyword.has_key?(cmd, key) do
      true -> key
      false -> nil
    end
  end

  defp pulling_out(cmd, flags, key) when is_atom(key) do
    docs = Keyword.get(cmd, key)
    cmd = Keyword.delete(cmd, key)

    payload_0 = section(payload_type: 0, payload: payload(doc: cmd))
    payload_1 = section(payload_type: 1, payload: payload(sequence: sequence(identifier: to_string(key), docs: docs)))

    op_msg(flags: flags, sections: [payload_0, payload_1])
  end

  defp flags(flags) do
    Enum.reduce(flags, [], fn
      {flag, true}, acc -> [flag | acc]
      {_flag, false}, acc -> acc
    end)
  end

  defp handshake_command(%{stable_api: nil}, client) do
    [ismaster: 1, helloOk: true, client: client]
  end

  defp handshake_command(%{stable_api: stable_api}, client) do
    [client: client]
    |> StableVersion.merge_stable_api(stable_api)
    |> Keyword.put(:hello, 1)
  end

  defp hello_command(cmd, %{hello_ok: false}) do
    cmd
    |> Keyword.put(:helloOk, true)
    |> Keyword.put(:ismaster, 1)
  end

  defp hello_command(cmd, %{hello_ok: true, stable_api: stable_api}) do
    cmd
    |> StableVersion.merge_stable_api(stable_api)
    |> Keyword.put(:hello, 1)
  end
end