postgrex

defmodule Postgrex do
  @moduledoc """
  PostgreSQL driver for Elixir.

  This module handles the connection to Postgres, providing support
  for queries, transactions, connection backoff, logging, pooling and
  more.

  Note that the notifications API (pub/sub) supported by Postgres is
  handled by `Postgrex.Notifications`. Hence, to use this feature,
  you need to start a separate (notifications) connection.
  """

  alias Postgrex.Query

  @typedoc """
  A connection process name, pid or reference.

  A connection reference is used when making multiple requests to the same
  connection, see `transaction/3`.
  """
  @type conn :: DBConnection.conn

  @pool_timeout 5000
  @timeout 15_000
  @max_rows 500

  ### PUBLIC API ###

  @doc """
  Start the connection process and connect to postgres.

  ## Options

    * `:hostname` - Server hostname (default: PGHOST env variable, then localhost);
    * `:port` - Server port (default: PGPORT env variable, then 5432);
    * `:database` - Database (required);
    * `:username` - Username (default: PGUSER env variable, then USER env var);
    * `:password` - User password (default PGPASSWORD);
    * `:parameters` - Keyword list of connection parameters;
    * `:timeout` - Connect timeout in milliseconds (default: `#{@timeout}`);
    * `:ssl` - Set to `true` if ssl should be used (default: `false`);
    * `:ssl_opts` - A list of ssl options, see ssl docs;
    * `:socket_options` - Options to be given to the underlying socket;
    * `:extensions` - A list of `{module, opts}` pairs where `module` is
    implementing the `Postgrex.Extension` behaviour and `opts` are the
    extension options;
    * `:decode_binary` - Either `:copy` to copy binary values when decoding with
    default extensions that return binaries or `:reference` to use a reference
    counted binary of the binary received from the socket. Referencing a
    potentially larger binary can be more efficient if the binary value is going
    to be garbaged collected soon because a copy is avoided. However the larger
    binary can not be garbage collected until all references are garbage
    collected (defaults to `:copy`);
    * `:prepare` - How to prepare queries, either `:named` to use named queries
    or `:unnamed` to force unnamed queries (default: `:named`);
    * `:transactions` - Set to `:strict` to error on unexpected transaction
    state, otherwise set to `naive` (default: `:naive`);
    * `:pool` - The pool module to use, see `DBConnection` for pool dependent
    options, this option must be included with all requests contacting the pool
    if not `DBConnection.Connection` (default: `DBConnection.Connection`);
    * `:null` - The atom to use as a stand in for postgres' `NULL` in encoding
    and decoding (default: `nil`);

  `Postgrex` uses the `DBConnection` framework and supports all `DBConnection`
  options. See `DBConnection` for more information.
  """
  @spec start_link(Keyword.t) :: {:ok, pid} | {:error, Postgrex.Error.t | term}
  def start_link(opts) do
    opts = [types: true] ++ Postgrex.Utils.default_opts(opts)
    DBConnection.start_link(Postgrex.Protocol, opts)
  end

  @doc """
  Runs an (extended) query and returns the result as `{:ok, %Postgrex.Result{}}`
  or `{:error, %Postgrex.Error{}}` if there was an error. Parameters can be
  set in the query as `$1` embedded in the query string. Parameters are given as
  a list of elixir values. See the README for information on how Postgrex
  encodes and decodes Elixir values by default. See `Postgrex.Result` for the
  result data.

  ## Options

    * `:pool_timeout` - Time to wait in the queue for the connection
    (default: `#{@pool_timeout}`)
    * `:queue` - Whether to wait for connection in a queue (default: `true`);
    * `:timeout` - Query request timeout (default: `#{@timeout}`);
    * `:decode_mapper` - Fun to map each row in the result to a term after
    decoding, (default: `fn x -> x end`);
    * `:pool` - The pool module to use, must match that set on
    `start_link/1`, see `DBConnection`
    * `:null` - The atom to use as a stand in for postgres' `NULL` in encoding
    and decoding;
    * `:mode` - set to `:savepoint` to use a savepoint to rollback to before the
    query on error, otherwise set to `:transaction` (default: `:transaction`);
    * `:copy_data` - Whether to add copy data as a final parameter for use
    with `COPY .. FROM STDIN` queries, if the query is not copying to the
    database the data is sent but silently discarded (default: `false`);

  ## Examples

      Postgrex.query(conn, "CREATE TABLE posts (id serial, title text)", [])

      Postgrex.query(conn, "INSERT INTO posts (title) VALUES ('my title')", [])

      Postgrex.query(conn, "SELECT title FROM posts", [])

      Postgrex.query(conn, "SELECT id FROM posts WHERE title like $1", ["%my%"])

      Postgrex.query(conn, "COPY posts TO STDOUT", [])

      Postgrex.query(conn, "COPY ints FROM STDIN", ["1\n2\n"], [copy_data: true])
  """
  @spec query(conn, iodata, list, Keyword.t) :: {:ok, Postgrex.Result.t} | {:error, Postgrex.Error.t}
  def query(conn, statement, params, opts \\ []) do
    query = %Query{name: "", statement: statement}
    case DBConnection.prepare_execute(conn, query, params, defaults(opts)) do
      {:ok, _, result} ->
        {:ok, result}
      {:error, %ArgumentError{} = err} ->
        raise err
      {:error, %RuntimeError{} = err} ->
        raise err
      {:error, _} = error ->
        error
    end
  end

  @doc """
  Runs an (extended) query and returns the result or raises `Postgrex.Error` if
  there was an error. See `query/3`.
  """
  @spec query!(conn, iodata, list, Keyword.t) :: Postgrex.Result.t
  def query!(conn, statement, params, opts \\ []) do
    query = %Query{name: "", statement: statement}
    {_, result} = DBConnection.prepare_execute!(conn, query, params, defaults(opts))
    result
  end

  @doc """
  Prepares an (extended) query and returns the result as
  `{:ok, %Postgrex.Query{}}` or `{:error, %Postgrex.Error{}}` if there was an
  error. Parameters can be set in the query as `$1` embedded in the query
  string. To execute the query call `execute/4`. To close the prepared query
  call `close/3`. See `Postgrex.Query` for the query data.

  ## Options

    * `:pool_timeout` - Time to wait in the queue for the connection
    (default: `#{@pool_timeout}`)
    * `:queue` - Whether to wait for connection in a queue (default: `true`);
    * `:timeout` - Prepare request timeout (default: `#{@timeout}`);
    * `:pool` - The pool module to use, must match that set on
    `start_link/1`, see `DBConnection`
    * `:null` - The atom to use as a stand in for postgres' `NULL` in encoding
    and decoding;
    * `:mode` - set to `:savepoint` to use a savepoint to rollback to before the
    prepare on error, otherwise set to `:transaction` (default: `:transaction`);
    * `:copy_data` - Whether to add copy data as the final parameter for use
    with `COPY .. FROM STDIN` queries, if the query is not copying to the
    database then the data is sent but ignored (default: `false`);

  ## Examples

      Postgrex.prepare(conn, "CREATE TABLE posts (id serial, title text)")
  """
  @spec prepare(conn, iodata, iodata, Keyword.t) :: {:ok, Postgrex.Query.t} | {:error, Postgrex.Error.t}
  def prepare(conn, name, statement, opts \\ []) do
    query = %Query{name: name, statement: statement}
    case DBConnection.prepare(conn, query, defaults(opts)) do
      {:error, %ArgumentError{} = err} ->
        raise err
      {:error, %RuntimeError{} = err} ->
        raise err
      other ->
        other
    end
  end

  @doc """
  Prepares an (extended) query and returns the prepared query or raises
  `Postgrex.Error` if there was an error. See `prepare/4`.
  """
  @spec prepare!(conn, iodata, iodata, Keyword.t) :: Postgrex.Query.t
  def prepare!(conn, name, statement, opts \\ []) do
    DBConnection.prepare!(conn, %Query{name: name, statement: statement}, defaults(opts))
  end

  @doc """
  Runs an (extended) prepared query and returns the result as
  `{:ok, %Postgrex.Result{}}` or `{:error, %Postgrex.Error{}}` if there was an
  error. Parameters are given as part of the prepared query, `%Postgrex.Query{}`.
  See the README for information on how Postgrex encodes and decodes Elixir
  values by default. See `Postgrex.Query` for the query data and
  `Postgrex.Result` for the result data.

  ## Options

    * `:pool_timeout` - Time to wait in the queue for the connection
    (default: `#{@pool_timeout}`)
    * `:queue` - Whether to wait for connection in a queue (default: `true`);
    * `:timeout` - Execute request timeout (default: `#{@timeout}`);
    * `:decode_mapper` - Fun to map each row in the result to a term after
    decoding, (default: `fn x -> x end`);
    * `:pool` - The pool module to use, must match that set on
    `start_link/1`, see `DBConnection`
    * `:mode` - set to `:savepoint` to use a savepoint to rollback to before the
    execute on error, otherwise set to `:transaction` (default: `:transaction`);

  ## Examples

      query = Postgrex.prepare!(conn, "CREATE TABLE posts (id serial, title text)")
      Postgrex.execute(conn, query, [])

      query = Postgrex.prepare!(conn, "SELECT id FROM posts WHERE title like $1")
      Postgrex.execute(conn, query, ["%my%"])
  """
  @spec execute(conn, Postgrex.Query.t, list, Keyword.t) ::
    {:ok, Postgrex.Result.t} | {:error, Postgrex.Error.t}
  def execute(conn, query, params, opts \\ []) do
    case DBConnection.execute(conn, query, params, defaults(opts)) do
      {:error, %ArgumentError{} = err} ->
        raise err
      {:error, %RuntimeError{} = err} ->
        raise err
      other ->
        other
    end
  end

  @doc """
  Runs an (extended) prepared query and returns the result or raises
  `Postgrex.Error` if there was an error. See `execute/4`.
  """
  @spec execute!(conn, Postgrex.Query.t, list, Keyword.t) :: Postgrex.Result.t
  def execute!(conn, query, params, opts \\ []) do
    DBConnection.execute!(conn, query, params, defaults(opts))
  end

  @doc """
  Closes an (extended) prepared query and returns `:ok` or
  `{:error, %Postgrex.Error{}}` if there was an error. Closing a query releases
  any resources held by postgresql for a prepared query with that name. See
  `Postgrex.Query` for the query data.

  ## Options

    * `:pool_timeout` - Time to wait in the queue for the connection
    (default: `#{@pool_timeout}`)
    * `:queue` - Whether to wait for connection in a queue (default: `true`);
    * `:timeout` - Close request timeout (default: `#{@timeout}`);
    * `:pool` - The pool module to use, must match that set on
    `start_link/1`, see `DBConnection`
    * `:mode` - set to `:savepoint` to use a savepoint to rollback to before the
    close on error, otherwise set to `:transaction` (default: `:transaction`);

  ## Examples

      query = Postgrex.prepare!(conn, "CREATE TABLE posts (id serial, title text)")
      Postgrex.close(conn, query)
  """
  @spec close(conn, Postgrex.Query.t, Keyword.t) :: :ok | {:error, Postgrex.Error.t}
  def close(conn, query, opts \\ []) do
    case DBConnection.close(conn, query, defaults(opts)) do
      {:ok, _} ->
        :ok
      {:error, %ArgumentError{} = err} ->
        raise err
      {:error, %RuntimeError{} = err} ->
        raise err
      {:error, _} = error ->
        error
    end
  end

  @doc """
  Closes an (extended) prepared query and returns `:ok` or raises
  `Postgrex.Error` if there was an error. See `close/3`.
  """
  @spec close!(conn, Postgrex.Query.t, Keyword.t) :: :ok
  def close!(conn, query, opts \\ []) do
    DBConnection.close!(conn, query, defaults(opts))
  end

  @doc """
  Acquire a lock on a connection and run a series of requests inside a
  transaction. The result of the transaction fun is return inside an `:ok`
  tuple: `{:ok, result}`.

  To use the locked connection call the request with the connection
  reference passed as the single argument to the `fun`. If the
  connection disconnects all future calls using that connection
  reference will fail.

  `rollback/2` rolls back the transaction and causes the function to
  return `{:error, reason}`.

  `transaction/3` can be nested multiple times if the connection
  reference is used to start a nested transaction. The top level
  transaction function is the actual transaction.

  ## Options

    * `:pool_timeout` - Time to wait in the queue for the connection
    (default: `#{@pool_timeout}`)
    * `:queue` - Whether to wait for connection in a queue (default: `true`);
    * `:timeout` - Transaction timeout (default: `#{@timeout}`);
    * `:pool` - The pool module to use, must match that set on
    `start_link/1`, see `DBConnection`;
    * `:mode` - Set to `:savepoint` to use savepoints instead of an SQL
    transaction, otherwise set to `:transaction` (default: `:transaction`);


  The `:timeout` is for the duration of the transaction and all nested
  transactions and requests. This timeout overrides timeouts set by internal
  transactions and requests. The `:pool` and `:mode` will be used for all
  requests inside the transaction function.

  ## Example

      {:ok, res} = Postgrex.transaction(pid, fn(conn) ->
        Postgrex.query!(conn, "SELECT title FROM posts", [])
      end)
  """
  @spec transaction(conn, ((DBConnection.t) -> result), Keyword.t) ::
    {:ok, result} | {:error, any} when result: var
  def transaction(conn, fun, opts \\ []) do
    DBConnection.transaction(conn, fun, defaults(opts))
  end

  @doc """
  Rollback a transaction, does not return.

  Aborts the current transaction fun. If inside multiple `transaction/3`
  functions, bubbles up to the top level.

  ## Example

      {:error, :oops} = Postgrex.transaction(pid, fn(conn) ->
        DBConnection.rollback(conn, :bar)
        IO.puts "never reaches here!"
      end)
  """
  @spec rollback(DBConnection.t, any) :: no_return()
  defdelegate rollback(conn, any), to: DBConnection

  @doc """
  Returns a cached map of connection parameters.

  ## Options

    * `:pool_timeout` - Call timeout (default: `#{@pool_timeout}`)
    * `:pool` - The pool module to use, must match that set on
    `start_link/1`, see `DBConnection`

  """
  @spec parameters(conn, Keyword.t) :: %{binary => binary}
  def parameters(conn, opts \\ []) do
    DBConnection.execute!(conn, %Postgrex.Parameters{}, nil, defaults(opts))
  end

  @doc """
  Returns a supervisor child specification for a DBConnection pool.
  """
  @spec child_spec(Keyword.t) :: Supervisor.Spec.spec
  def child_spec(opts) do
    opts = [types: true] ++ Postgrex.Utils.default_opts(opts)
    DBConnection.child_spec(Postgrex.Protocol, opts)
  end

  @doc """
  Returns a stream for a prepared query on a connection.

  Stream consumes memory in chunks of at most `max_rows` rows (see Options).
  This is useful for processing _large_ datasets.

  A stream must be wrapped in a transaction and may be used as an `Enumerable`
  or a `Collectable`.

  When used as an `Enumerable` with a `COPY .. TO STDOUT` SQL query no other
  queries or streams can be interspersed until the copy has finished. Otherwise
  it is possible to intersperse enumerable streams and queries.

  When used as a `Collectable` the query must have been prepared with
  `copy_data: true`, otherwise it will raise. Instead of using an extra
  parameter for the copy data, the data from the enumerable is copied to the
  database. No other queries or streams can be interspersed until the copy has
  finished. If the query is not copying to the database the copy data will still
  be sent but is silently discarded.

  ### Options

    * `:max_rows` - Maximum numbers of rows in a result (default to `#{@max_rows}`)
    * `:decode_mapper` - Fun to map each row in the result to a term after
    decoding, (default: `fn x -> x end`);
    * `:mode` - set to `:savepoint` to use a savepoint to rollback to before an
    execute on error, otherwise set to `:transaction` (default: `:transaction`);

  ## Examples

      Postgrex.transaction(pid, fn(conn) ->
        query = Postgrex.prepare!(conn, "COPY posts TO STDOUT")
        stream = Postgrex.stream(conn, query, [])
        Enum.into(stream, File.stream!("posts"))
      end)

      Postgrex.transaction(pid, fn(conn) ->
        query = Postgrex.prepare!(conn, "COPY posts FROM STDIN", [copy_data: true])
        stream = Postgrex.stream(conn, query, [])
        Enum.into(File.stream!("posts"), stream)
      end)
  """
  @spec stream(DBConnection.t, Postgrex.Query.t, list, Keyword.t) :: Postgrex.Stream.t
  def stream(%DBConnection{} = conn, query, params, options \\ [])  do
    max_rows = options[:max_rows] || @max_rows
    %Postgrex.Stream{conn: conn, max_rows: max_rows, options: options,
                     params: params, query: query}
  end

  ## Helpers
  defp defaults(opts) do
    Keyword.put_new(opts, :timeout, @timeout)
  end
end