class Sequel::Postgres::Database

def check_database_errors
  begin
    yield
  rescue => e
    raise_error(e, :classes=>CONVERTED_EXCEPTIONS)
  end
end

def connect(server)
  opts = server_opts(server)
  conn = if SEQUEL_POSTGRES_USES_PG
    connection_params = {
      :host => opts[:host],
      :port => opts[:port] || 5432,
      :dbname => opts[:database],
      :user => opts[:user],
      :password => opts[:password],
      :connect_timeout => opts[:connect_timeout] || 20,
      :sslmode => opts[:sslmode]
    }.delete_if { |key, value| blank_object?(value) }
    Adapter.connect(connection_params)
  else
    Adapter.connect(
      (opts[:host] unless blank_object?(opts[:host])),
      opts[:port] || 5432,
      nil, '',
      opts[:database],
      opts[:user],
      opts[:password]
    )
  end
  if encoding = opts[:encoding] || opts[:charset]
    if conn.respond_to?(:set_client_encoding)
      conn.set_client_encoding(encoding)
    else
      conn.async_exec("set client_encoding to '#{encoding}'")
    end
  end
  conn.db = self
  conn.apply_connection_settings
  @conversion_procs ||= get_conversion_procs(conn)
  conn
end

def copy_table(table, opts={})
  sql = if table.is_a?(String)
    sql = table
  else
    if opts[:options] || opts[:format]
      options = " ("
      options << "FORMAT #{opts[:format]}" if opts[:format]
      options << "#{', ' if opts[:format]}#{opts[:options]}" if opts[:options]
      options << ')'
    end
    table = if table.is_a?(::Sequel::Dataset)
      "(#{table.sql})"
    else
      literal(table)
    end
   sql = "COPY #{table} TO STDOUT#{options}"
  end
  synchronize(opts[:server]) do |conn| 
    conn.execute(sql)
    begin
      if block_given?
        while buf = conn.get_copy_data
          yield buf
        end
        nil
      else
        b = ''
        b << buf while buf = conn.get_copy_data
        b
      end
    ensure
      raise DatabaseDisconnectError, "disconnecting as a partial COPY may leave the connection in an unusable state" if buf
    end
  end 
end

def disconnect_connection(conn)
  begin
    conn.finish
  rescue PGError
  end
end

def execute(sql, opts={}, &block)
  check_database_errors do
    return execute_prepared_statement(sql, opts, &block) if Symbol === sql
    synchronize(opts[:server]){|conn| conn.execute(sql, opts[:arguments], &block)}
  end
end

def execute_insert(sql, opts={})
  return execute(sql, opts) if Symbol === sql
  check_database_errors do
    synchronize(opts[:server]) do |conn|
      conn.execute(sql, opts[:arguments])
      insert_result(conn, opts[:table], opts[:values])
    end
  end
end

def execute_prepared_statement(name, opts={})
  ps = prepared_statements[name]
  sql = ps.prepared_sql
  ps_name = name.to_s
  args = opts[:arguments]
  synchronize(opts[:server]) do |conn|
    unless conn.prepared_statements[ps_name] == sql
      if conn.prepared_statements.include?(ps_name)
        conn.execute("DEALLOCATE #{ps_name}") unless conn.prepared_statements[ps_name] == sql
      end
      conn.prepared_statements[ps_name] = sql
      conn.check_disconnect_errors{log_yield("PREPARE #{ps_name} AS #{sql}"){conn.prepare(ps_name, sql)}}
    end
    q = conn.check_disconnect_errors{log_yield("EXECUTE #{ps_name}", args){conn.exec_prepared(ps_name, args)}}
    if opts[:table] && opts[:values]
      insert_result(conn, opts[:table], opts[:values])
    else
      begin
        block_given? ? yield(q) : q.cmd_tuples
      ensure
        q.clear
      end
    end
  end
end

def get_conversion_procs(conn)
  procs = PG_TYPES.dup
  procs[1114] = method(:to_application_timestamp)
  procs[1184] = method(:to_application_timestamp)
  conn.execute("SELECT oid, typname FROM pg_type where typtype = 'b'") do |res|
    res.ntuples.times do |recnum|
      if pr = PG_NAMED_TYPES[res.getvalue(recnum, 1).untaint.to_sym]
        procs[res.getvalue(recnum, 0).to_i] ||= pr
      end
    end
  end
  procs
end

def initialize(*args)
  super
  @primary_keys = {}
  @primary_key_sequences = {}
end

def listen(channels, opts={}, &block)
  check_database_errors do
    synchronize(opts[:server]) do |conn|
      begin
        channels = Array(channels)
        channels.each{|channel| conn.execute("LISTEN #{channel}")}
        opts[:after_listen].call(conn) if opts[:after_listen]
        timeout = opts[:timeout] ? [opts[:timeout]] : []
        if l = opts[:loop]
          raise Error, 'calling #listen with :loop requires a block' unless block
          loop_call = l.respond_to?(:call)
          catch(:stop) do
            loop do
              conn.wait_for_notify(*timeout, &block)
              l.call(conn) if loop_call
            end
          end
          nil
        else
          conn.wait_for_notify(*timeout, &block)
        end
      ensure
        conn.execute("UNLISTEN *")
      end
    end
  end
end