snowpack

defmodule Snowpack.ODBC do
  @moduledoc """
  Adapter to Erlang's `:odbc` module.

  This module is a GenServer that handles communication between Elixir
  and Erlang's `:odbc` module.

  It is used by `Snowpack.Protocol` and should not generally be
  accessed directly.
  """

  use GenServer

  alias Snowpack.Error

  require Logger

  @default_query_timeout :timer.seconds(120)
  @begin_transaction 'begin transaction;'
  @last_query_id 'SELECT LAST_QUERY_ID() as query_id;'
  @close_transaction 'commit;'

  @data_types [
    {~r/NUMBER\([0-9]+,0\)/, :integer},
    {~r/DECIMAL\([0-9]+,0\)/, :integer},
    {~r/NUMERIC\([0-9]+,0\)/, :integer},
    {~r/INT\([0-9]+,0\)/, :integer},
    {~r/INTEGER\([0-9]+,0\)/, :integer},
    {~r/BIGINT\([0-9]+,0\)/, :integer},
    {~r/SMALLINT\([0-9]+,0\)/, :integer},
    {~r/NUMBER\([0-9]+,[1-9]+\)/, :float},
    {~r/DECIMAL\([0-9]+,[1-9]+\)/, :float},
    {~r/NUMERIC\([0-9]+,[1-9]+\)/, :float},
    {~r/INT\([0-9]+,[1-9]+\)/, :float},
    {~r/INTEGER\([0-9]+,[1-9]+\)/, :float},
    {~r/BIGINT\([0-9]+,[1-9]+\)/, :float},
    {~r/SMALLINT\([0-9]+,[1-9]+\)/, :float},
    {~r/BOOLEAN/, :boolean},
    {~r/FLOAT/, :float},
    {~r/DOUBLE/, :float},
    {~r/REAL/, :float},
    {~r/DATETIME/, :datetime},
    {~r/TIMESTAMP/, :datetime},
    {~r/DATE/, :date},
    {~r/TIME/, :time},
    {~r/OBJECT/, :json},
    {~r/ARRAY/, :array},
    {~r/VARIANT/, :variant}
  ]

  ## Public API

  @doc """
  Starts the connection process to the ODBC driver.

  `conn_str` should be a connection string in the format required by your ODBC driver.
  `opts` will be passed verbatim to `:odbc.connect/2`.
  """
  @spec start_link(Keyword.t()) :: {:ok, pid()}
  def start_link(opts) do
    GenServer.start_link(__MODULE__, opts)
  end

  @doc """
  Sends a parametrized query to the ODBC driver.

  Interface to `:odbc.param_query/3`.See
  [Erlang's ODBC guide](http://erlang.org/doc/apps/odbc/getting_started.html)
  for usage details and examples.

  * `pid` is the `:odbc` process id
  * `statement` is the SQL query string
  * `params` are the parameters to send with the SQL query
  * `opts` are options to be passed on to `:odbc`
    * `timeout` in millisecods (defaults to 2 minutes)
  * `with_query_id` runs query in transaction and selects LAST_QUERY_ID()
  """
  @spec query(pid(), iodata(), Keyword.t(), Keyword.t(), boolean()) ::
          {:selected, [binary()], [tuple()]}
          | {:selected, [binary()], [tuple()], [{binary()}]}
          | {:updated, non_neg_integer()}
          | {:error, Error.t()}
  def query(pid, statement, params, opts, with_query_id \\ false) do
    if Process.alive?(pid) do
      statement = IO.iodata_to_binary(statement)

      timeout = Keyword.get(opts, :timeout, @default_query_timeout)

      # call timeout is query timeout + 60s buffer
      call_timeout = timeout + :timer.seconds(60)

      GenServer.call(
        pid,
        {:query, %{statement: statement, params: params, with_query_id: with_query_id, timeout: timeout}},
        call_timeout
      )
    else
      {:error, %Error{message: :no_connection}}
    end
  end

  @spec describe_result(pid(), iodata()) ::
          {:selected, [binary()], [tuple()]}
          | {:error, Error.t()}
  def describe_result(pid, query_id) do
    if Process.alive?(pid) do
      GenServer.call(pid, {:describe_result, query_id})
    else
      {:error, %Error{message: :no_connection}}
    end
  end

  @doc """
  Disconnects from the ODBC driver.

  `pid` is the `:odbc` process id
  """
  # TODO: figure out how to test this correctly
  # coveralls-ignore-start
  @spec disconnect(pid()) :: :ok
  def disconnect(pid) do
    GenServer.stop(pid, :normal)
  end

  # coveralls-ignore-stop

  ## GenServer callbacks

  @spec init(keyword) :: {:ok, any}
  def init(opts) do
    send(self(), {:start, opts})
    {:ok, %{backoff: :backoff.init(2, 60), state: :not_connected}}
  end

  @spec handle_call(request :: term(), term(), state :: term()) ::
          {:noreply, term()}
          | {:noreply, term(), :hibernate | :infinity | non_neg_integer() | {:continue, term()}}
          | {:reply, term(), term()}
          | {:stop, term(), term()}
          | {:reply, term(), term(), :hibernate | :infinity | non_neg_integer() | {:continue, term()}}
          | {:stop, term(), term(), term()}
  def handle_call({:query, _query}, _from, %{state: :not_connected} = state) do
    {:reply, {:error, :not_connected}, state}
  end

  def handle_call(
        {:query, %{statement: statement, params: params, with_query_id: false, timeout: timeout}},
        _from,
        %{pid: pid} = state
      ) do
    case :odbc.param_query(pid, :binary.bin_to_list(statement), params, timeout) do
      {:error, reason} ->
        error = Error.exception(reason)
        Logger.warn("Unable to execute query: #{error.message}")

        {:reply, {:error, error}, state}

      result ->
        {:reply, result, state}
    end
  end

  def handle_call(
        {:query, %{statement: statement, params: params, with_query_id: true, timeout: timeout}},
        _from,
        %{pid: pid} = state
      ) do
    :odbc.sql_query(pid, @begin_transaction)

    case :odbc.param_query(pid, :binary.bin_to_list(statement), params, timeout) do
      {:error, reason} ->
        error = Error.exception(reason)
        Logger.warn("Unable to execute query: #{error.message}")

        :odbc.sql_query(pid, @close_transaction)

        {:reply, {:error, error}, state}

      result ->
        {:selected, _, query_id} = :odbc.sql_query(pid, @last_query_id)

        :odbc.sql_query(pid, @close_transaction)

        {:reply, Tuple.append(result, query_id), state}
    end
  end

  def handle_call({:describe_result, query_id}, _from, %{pid: pid} = state) do
    case :odbc.sql_query(pid, :binary.bin_to_list("DESCRIBE RESULT '#{query_id}'")) do
      {:error, reason} ->
        # coveralls-ignore-start
        error = Error.exception(reason)
        Logger.warn("Unable to describe #{query_id}: #{error.message}")

        case error.odbc_code do
          "02000" ->
            Logger.warn("Query #{query_id} has expired.")
            {:reply, [], state}

          _ ->
            {:reply, {:error, error}, state}
        end

      # coveralls-ignore-stop

      result ->
        # parse name, type
        {:reply, build_type_tuples(result), state}
    end
  end

  @spec handle_info(msg :: :timeout | term(), state :: term()) ::
          {:noreply, term()}
          | {:noreply, term(), :hibernate | :infinity | non_neg_integer() | {:continue, term()}}
          | {:stop, term(), term()}
  def handle_info({:start, opts}, %{backoff: backoff} = _state) do
    connect_opts =
      opts
      |> Keyword.delete_first(:conn_str)
      |> Keyword.put_new(:auto_commit, :on)
      |> Keyword.put_new(:binary_strings, :on)
      |> Keyword.put_new(:tuple_row, :on)
      |> Keyword.put_new(:extended_errors, :on)

    case :odbc.connect(opts[:conn_str], connect_opts) do
      {:ok, pid} ->
        {:noreply, %{pid: pid, backoff: :backoff.succeed(backoff), state: :connected}}

      {:error, reason} ->
        # coveralls-ignore-start
        error = Error.exception(reason)

        Logger.warn("Unable to connect to snowflake: #{error.message}")

        Process.send_after(
          self(),
          {:start, opts},
          backoff |> :backoff.get() |> :timer.seconds()
        )

        {_, new_backoff} = :backoff.fail(backoff)
        {:noreply, %{backoff: new_backoff, state: :not_connected}}
        # coveralls-ignore-stop
    end
  end

  # TODO: figure out how to test this correctly
  # coveralls-ignore-start
  @spec terminate(term(), state :: term()) :: term()
  def terminate(_reason, %{state: :not_connected} = _state), do: :ok
  def terminate(_reason, %{pid: pid} = _state), do: :odbc.disconnect(pid)
  # coveralls-ignore-stop

  defp build_type_tuples({_, columns, rows}) do
    Enum.map(rows, fn row ->
      columns
      |> Enum.zip_reduce(Tuple.to_list(row), %{}, fn name, value, acc ->
        Map.put(acc, to_string(name), value)
      end)
      |> Kernel.then(&data_type_from_column_metadata/1)
    end)
  end

  defp data_type_from_column_metadata(%{"name" => name, "type" => col_type}) do
    {_, type} =
      Enum.find(@data_types, {nil, :default}, fn {reg, _type} ->
        String.match?(col_type, reg)
      end)

    {name, type}
  end
end