require 'flydata/source/generate_source_dump'
require 'flydata/preference/data_entry_preference'
require 'flydata/source_postgresql/postgresql_component'
require 'flydata/source_postgresql/query_based_sync/diff_query_generator'
require 'flydata/source_postgresql/table_meta'
require 'flydata-core/postgresql/source_pos'
require 'flydata-core/postgresql/pg_client'
require 'msgpack'

module Flydata
module SourcePostgresql

class GenerateSourceDump < Source::GenerateSourceDump
  include PostgresqlComponent

  def run_compatibility_check(dump_dir, backup_dir)
    %w(host username database).each do |k|
      if de_prefs[k].to_s.empty?
        raise "'#{k}' is required. Set the value in the conf file " +
              "-> #{Flydata::Preference::DataEntryPreference.conf_path(de)}"
      end
    end
  end

  def confirmation_items
    items = {
      "host"     => de_prefs['host'],
      "port"     => de_prefs['port'],
      "username" => de_prefs['username'],
      "database" => de_prefs['database'],
      "schema"   => de_prefs['schema'],
    }

    items
  end

  DUMP_SIZE_QUERY = <<EOS
SELECT sum(pg_total_relation_size(c.oid)) AS "total_size"
  FROM pg_class c
  LEFT JOIN pg_namespace n ON n.oid = c.relnamespace
  WHERE nspname = %{schema} AND relname in (%{tables})
EOS

  def dump_size(tables)
    cli = FlydataCore::Postgresql::PGClient.new(de_prefs)

    base_query = DUMP_SIZE_QUERY % {
      schema: FlydataCore::Postgresql::QueryHelper.schema_as_value(de_prefs['schema']),
      tables: '%s'   # Use binding parameters for tables
    }
    query = FlydataCore::Postgresql::PGQuery.new(base_query,
              placeholder_size: tables.size,
              placeholder_start_num: 1)
    res = cli.query(query, tables)

    res.first['total_size'].to_i

  ensure
    cli.close if cli
  end

  def dump(tables, file_path = nil, &src_pos_callback)
    io = nil
    if file_path
      io = File.open(file_path, "w")
    else
      raise "dump via pipe has not been implemented yet"
    end

    table_meta = Flydata::SourcePostgresql::TableMeta.new(de_prefs, tables)
    cli = FlydataCore::Postgresql::PGClient.new(de_prefs)
    table_meta.reload(cli)

    context = source.sync_generate_table_ddl(dp, nil)
    source_pos = get_source_pos(table_meta.current_snapshot, &src_pos_callback)

    options = de_prefs.merge(table_meta: table_meta)
    missing_tables = context.each_source_tabledef(tables, options) do |tabledef, error|
      dump_table(tabledef, source_pos, io, cli) if tabledef
    end

    nil
  ensure
    cli.close if cli
    io.close if io
  end

  private

  def get_source_pos(snapshot, &src_pos_callback)
    src_pos = FlydataCore::Postgresql::SourcePos.new(snapshot)
    src_pos_callback.call(nil, src_pos)

    src_pos
  end

  NUM_ROWS = 500000

  def dump_table(tabledef, source_pos, io, cli)
    dump_source_table(tabledef, io)

    last_pks = nil
    loop do
      num_rows, last_pks =
        dump_table_chunk(tabledef.table_name, de_prefs['schema'], source_pos, NUM_ROWS,
                         tabledef, tabledef.pk_columns, last_pks, io, cli)
      break if num_rows < NUM_ROWS
    end
  end

  def dump_source_table(tabledef, io)
    columns = {}
    tabledef.column_def.each do |column_name, columndef_str|
      columndef_hash = eval columndef_str
      columns[column_name] = {"column_name" => column_name,
                              "format_type" => columndef_hash["data_type"]}
    end
    source_table_hash = { "table_name" => tabledef.table_name,
                          "columns" => columns }

    io.write(source_table_hash.to_msgpack)
  end

  def dump_table_chunk(table, schema, source_pos, num_rows,
                       tabledef, pk_columns, last_pks, io, cli)
    query = Flydata::SourcePostgresql::QueryBasedSync::DiffQueryGenerator.new(
              table, schema,
              columns: tabledef.columns,
              to_sid: source_pos.snapshot,
              pk_columns: pk_columns,
              last_pks: last_pks,
              limit: num_rows).build_query

    column_names = tabledef.columns.inject([]) do |ary, h|
       ary << h[:column]
    end

    res = cli.query(query)
    count = 0
    last_row = nil
    res.each do |row|
      count += 1
      last_row = row
      dump_row(row, column_names, io)
    end

    last_pks = last_row ? pk_columns.collect{|col| last_row[col]} : nil

    [count, last_pks]
  end

  def dump_row(row, columns, io)
    data = columns.collect{|col| row[col] }.to_msgpack
    io.write(data)
  end
end

end
end