require 'date'
require 'flydata-core/errors'
require 'flydata-core/table_def/value_conv'

module FlydataCore
module TableDef

class NameTooLongError < RuntimeError; end

class RedshiftTableDef
  TYPE_MAP_F2R = {
    'binary' => {type: 'varchar', use_params: true, default_value: ''},
    'bit' => {type: 'bigint', default_value: '0'},
    'boolean' => {type: 'boolean', default_value: false},
    'varbit' => {type: 'bigint', default_value: '0'},
    'bytea' => {type: 'varchar(max)', default_value: ''},
    'char' => {type: 'char', use_params: true, default_value: ''},
    'date' => {type: 'date', default_value: '0000-01-01'},
    'datetime' => {type: 'timestamp', default_value: '0000-01-01'},
    'datetimetz' => {type: 'timestamp', default_value: '0000-01-01'},
    'enum' => {type: 'varchar encode bytedict', default_value: ''},
    'float4' => {type: 'float4', default_value: '0'},
    'float4 unsigned' => {type: 'float4', default_value: '0'},
    'float8' => {type: 'float8', default_value: '0'},
    'float8 unsigned' => {type: 'float8', default_value: '0'},
    'int1' => {type: 'int2', default_value: '0'},
    'int1 unsigned' => {type: 'int2', unsigned: true, default_value: '0'},
    'int2' => {type: 'int2', default_value: '0'},
    'int2 unsigned' => {type: 'int4', unsigned: true, default_value: '0'},
    'int3' => {type: 'int4', default_value: '0'},
    'int3 unsigned' => {type: 'int4', unsigned: true, default_value: '0'},
    'int4' => {type: 'int4', default_value: '0'},
    'int4 unsigned' => {type: 'int8', unsigned: true, default_value: '0'},
    'int8' => {type: 'int8', default_value: '0'},
    'int8 unsigned' => {type: 'numeric(20,0)', unsigned: true, default_value: '0'},
    'money' => {type: 'numeric', use_params: true, max_size: [38,37], default_value: '0'},
    'numeric' => {type: 'numeric', use_params: true, default_params: [18,8], max_size: [38,37], default_value: '0'},
    'numeric unsigned' => {type: 'numeric', use_params: true, default_params: [18,8], max_size: [38,37], default_value: '0'},
    'set' => {type: 'varchar encode bytedict', default_value: ''},
    'text' => {type: 'varchar(max)', default_value: ''},
    'time' => {type: 'timestamp', default_value: '0000-01-01'},
    'timetz' => {type: 'timestamp', default_value: '0000-01-01'},
    'varbinary' => {type: 'varchar', use_params: true, max_size: 65535, default_value: ''},
    'varchar' => {type: 'varchar', use_params: true, max_size: 65535, default_value: ''},
    'year' => {type: 'date', default_value: '0001-01-01'},
    '_unsupported' => {type: 'varchar(max)', default_value: ''},
  }
  def self.from_flydata_tabledef(flydata_tabledef, options = {})
    options[:flydata_ctl_table] = true unless options.has_key?(:flydata_ctl_table)
    schema_name = options[:schema_name]

    begin
      tabledef = ""
      tabledef += create_schema_sql(schema_name) if options[:flydata_ctl_table] && !schema_name.to_s.empty?
      tabledef += create_flydata_ctl_table_sql(schema_name) if options[:flydata_ctl_table]
      options[:backup_postfix] = "_flydata#{Time.now.strftime('%Y%m%d%H%M%S')}"
      tabledef += "BEGIN;\n"
      tabledef += remove_table_sql(flydata_tabledef, schema_name, options) unless options[:ctl_only]
      tabledef += create_table_sql(flydata_tabledef, schema_name, options) unless options[:ctl_only]
      tabledef += comment_sql(flydata_tabledef, schema_name) unless options[:ctl_only]
      tabledef += flydata_ctl_sql(flydata_tabledef, schema_name)
      tabledef += "COMMIT;\n"
      tabledef += drop_backup_table_sql(flydata_tabledef, schema_name, options) unless options[:ctl_only]
    rescue => e
      # Catch errors from generating schema. Generally an unsupported data type
      raise TableDefError, {error: "Errors generating schema. Please contact us for further instructions", table: flydata_tabledef[:table_name]}
    end
  end

  FLYDATA_CTL_COLUMNS_TABLE = "flydata_ctl_columns"
  FLYDATA_CTL_TABLES_TABLE = "flydata_ctl_tables"
  CREATE_SCHEMA_SQL = <<EOS
CREATE SCHEMA IF NOT EXISTS "%s";
EOS
  CREATE_FLYDATA_CTL_COLUMNS_SQL = <<EOS
CREATE TABLE IF NOT EXISTS %s(
  id integer NOT NULL IDENTITY(1,1),
  table_name varchar(128) NOT NULL,
  column_name varchar(128) NOT NULL,
  src_data_type varchar(1024) NOT NULL,
  revision int NOT NULL DEFAULT 1,
  ordinal_position int NOT NULL,
  PRIMARY KEY(id)
) DISTKEY(table_name) SORTKEY(table_name);
EOS
  CREATE_FLYDATA_CTL_TABLES_SQL = <<EOS
CREATE TABLE IF NOT EXISTS %s(
  id integer NOT NULL IDENTITY(1,1),
  table_name varchar(128) NOT NULL,
  attribute varchar(128) NOT NULL,
  value varchar(max),
  created_at timestamp DEFAULT SYSDATE,
  PRIMARY KEY(id)
) DISTKEY(table_name) SORTKEY(table_name);
EOS
  CREATE_FLYDATA_CTL_TABLE_SQL = "#{CREATE_FLYDATA_CTL_COLUMNS_SQL}#{CREATE_FLYDATA_CTL_TABLES_SQL}"


  def self.table_name_for_ddl(table_name, schema_name)
    schema_name.to_s.empty? ? "\"#{table_name}\"" : "\"#{schema_name}\".\"#{table_name}\""
  end

  def self.fq_table_name(table_name, schema_name)
    idx = table_name.rindex('.')
    if (idx)
      # table name has schame qualifier.  Remove it.
      table_name = table_name[idx+1..-1]
    end

    result =
      schema_name.to_s.empty? ? table_name : "#{schema_name}.#{table_name}"

    convert_to_valid_table_name(result)
  end

  def self.flydata_ctl_table_for_ddl(schema_name, ctl_table_type = :columns)
    table_name = case ctl_table_type
                 when :columns; FLYDATA_CTL_COLUMNS_TABLE
                 when :tables;  FLYDATA_CTL_TABLES_TABLE
                 end
    table_name_for_ddl(table_name, schema_name)
  end

  def self.create_schema_sql(schema_name)
    CREATE_SCHEMA_SQL % schema_name
  end

  def self.create_flydata_ctl_table_sql(schema_name)
    # No drop table here intentionally because losing the data is fatal.
    columns_tbl = flydata_ctl_table_for_ddl(schema_name, :columns)
    tables_tbl = flydata_ctl_table_for_ddl(schema_name, :tables)
    CREATE_FLYDATA_CTL_TABLE_SQL % [columns_tbl, tables_tbl]
  end

  def self.create_flydata_ctl_tables_sql(schema_name)
    tables_tbl = flydata_ctl_table_for_ddl(schema_name, :tables)
    CREATE_FLYDATA_CTL_TABLES_SQL % [tables_tbl]
  end

  DROP_TABLE_SQL = <<EOS
DROP TABLE IF EXISTS %s;
EOS

  RENAME_TABLE_SQL = <<EOS
ALTER TABLE %s RENAME TO %s;
EOS

  CREATE_TABLE_SQL = <<EOS
CREATE TABLE %s (
%s
)%s;
EOS

  CREATE_TABLE_IF_NOT_EXISTS_SQL = <<EOS
CREATE TABLE IF NOT EXISTS %s (
%s
)%s;
EOS

  def self.remove_table_sql(flydata_tabledef, schema_name, options)
    return "" if options[:skip_drop_table]
    table_name = flydata_tabledef[:table_name]
    redshift_tbl = table_name_for_ddl(table_name, schema_name)
    sql = ""
    if options[:backup_postfix]
      # drop backup table if exists
      sql += drop_backup_table_sql(flydata_tabledef, schema_name, options)
      # create an empty table to prevent RENAME TABLE query from failing
      sql += create_table_sql(flydata_tabledef, schema_name,
                       options.merge(create_table_sql: CREATE_TABLE_IF_NOT_EXISTS_SQL))
      backup_tbl = table_name_for_ddl(
                     "#{table_name}#{options[:backup_postfix]}", nil)
      sql += RENAME_TABLE_SQL % [redshift_tbl, backup_tbl]
    else
      sql += DROP_TABLE_SQL % redshift_tbl
    end
    sql
  end

  def self.drop_backup_table_sql(flydata_tabledef, schema_name, options)
    return "" if options[:skip_drop_table]
    sql = ""
    if options[:backup_postfix]
      table_name = flydata_tabledef[:table_name]
      backup_tbl = table_name_for_ddl(
                     "#{table_name}#{options[:backup_postfix]}", schema_name)
      sql += DROP_TABLE_SQL % backup_tbl
    end

    sql
  end

  def self.create_table_sql(flydata_tabledef, schema_name, options = {})
    ct_sql = options[:create_table_sql] || CREATE_TABLE_SQL
    lines = flydata_tabledef[:columns].collect{|column| column_def_sql(column) }
    pk_def = primary_key_sql(flydata_tabledef)

    unless options[:skip_primary_key_check]
      unless pk_def
        $log.error "No primary key defined in table: #{flydata_tabledef[:table_name]}"
        raise "No primary key defined"
      end
    end
    lines << pk_def if pk_def

    contents = lines.join(",\n")

    dk_sk_def = distkey_sortkey_sql(flydata_tabledef)

    table_name = flydata_tabledef[:table_name]
    redshift_tbl = table_name_for_ddl(table_name, schema_name)
    sql = ""
    sql += ct_sql % [redshift_tbl, contents, dk_sk_def]
    sql
  end

  def self.column_def_sql(column, opt = {})
    type = column[:type]
    if type =~ /\((.*?)\)/
      type = $` + $'
      params = $1
    end

    type_info = TYPE_MAP_F2R[type]
    raise "Unsupported type '#{column[:type]}'" if type_info.nil?

    rs_type = if type_info[:use_params]
                if params
                  params = check_and_replace_max(params, Array(type_info[:max_size]), type) if type_info[:max_size]
                else
                  #source data type has no parameter. use default parameters.
                  raise "No default pramameters for type:`#{column[:type]}`" unless type_info[:default_params]
                  params = type_info[:default_params].join(',')
                end
                type_info[:type] + "(#{params})"
              else
                type_info[:type]
              end
    line = %Q|  "#{convert_to_valid_column_name(column[:column])}" #{rs_type}|

    not_null = nil
#   not_null = column[:not_null]
    # `not_null` is always `nil`.
    # ***`column[:not_null]` is ignored for Redshift***
    # Avoid explicitly creating NOT NULL column in Redshift.
    # Otherwise, removing NOT NULL attribute from the column later would cause an error
    # because Redshift doesn't support 'ALTER TABLE tbl MODIFY column column_attributes'.

    line += " NOT NULL" if not_null

    if (column.has_key?(:default))
      val = replace_default_value(type, type_info[:type], column[:default])
      line += " DEFAULT #{val}" if val
    elsif not_null && opt[:for] == :alter_table
      # Redshift doesn't allow adding a not null column without default value
      # Add a defalt value
      line += " DEFAULT '#{type_info[:default_value]}'"
    end
# Commented out because no IDENTITY column must be used for a replicated table.
# Values come from the master.
#    line += " IDENTITY(1, 1)" if (column[:auto_increment])

    line
  end

  NULL_STR = "NULL"

  def self.replace_default_value(flydata_type, redshift_type, default_value)
    return NULL_STR if default_value.nil?
    # strip type cast
    if default_value.kind_of?(String) &&
       /^(.+?)(::"?[a-z ]+"?)*$/.match(default_value)
      default_value = $1
    end
    if flydata_type.start_with?('year')
      value = convert_year_into_date(remove_single_quote(default_value))
      begin
        Date.parse(value)
      rescue
        raise "default value of YEAR type must be 2 or 4-digit, value:'#{default_value}'"
      end
    elsif flydata_type.start_with?('money')
      default_value = ValueConv.strip_currency_format(remove_single_quote(default_value))
    end

    case redshift_type
    when 'timestamp'
      case default_value.upcase
      when /^CURRENT_TIMESTAMP\b/, /^CURRENT_TIME\b/, /^NOW\(\)/
        'SYSDATE'
      else
        "'#{self.parse_timestamp(remove_single_quote(default_value))}'"
      end
    when 'date'
      "'#{self.parse_date(remove_single_quote(default_value))}'"
    when 'boolean'
      default_value.to_s.upcase
    else
      if !default_value.kind_of?(String)
        "'#{default_value}'"
      elsif /^[Bb]'.+'$/.match(default_value)
        "0b#{default_value[2..-2]}".oct
      elsif /^[xX]'.+'$/.match(default_value)
        "0x#{default_value[2..-2]}".oct
      elsif /^0[bx].+/.match(default_value)
        default_value.oct
      elsif /\s*nextval\(/.match(default_value)
        # Redshift does not support nextval() function.  Set no default.
        nil
      elsif /^'.+'/.match(default_value)
        default_value
      else
        "'#{default_value}'"
      end
    end
  end

  def self.remove_single_quote(value)
    return value unless value.kind_of?(String)
    /^'.*'$/.match(value) ? value[1..-2] : value
  end

  def self.primary_key_sql(flydata_tabledef)
    pks = primary_keys(flydata_tabledef)
    pks.empty? ? nil : "  PRIMARY KEY (#{pks.collect{|pk| %Q|"#{pk}"|}.join(',')})"
  end

  def self.primary_keys(flydata_tabledef)
    flydata_tabledef[:columns].select{|col| col[:primary_key]}.collect{|col| convert_to_valid_column_name(col[:column])}
  end

  def self.distkey_sortkey_sql(flydata_tabledef)
    pks = primary_keys(flydata_tabledef)
    return nil if pks.empty?
    %Q| DISTKEY("#{pks.first}") SORTKEY(#{pks.collect{|pk| %Q|"#{pk}"|}.join(',')})|
  end

  def self.comment_sql(flydata_tabledef, schema_name)
    sql = ""
    flydata_tabledef[:columns].each do |col|
      next unless col[:comment]

      sql += <<EOS
COMMENT ON COLUMN #{table_name_for_ddl(flydata_tabledef[:table_name], schema_name)}."#{convert_to_valid_column_name(col[:column])}"
  IS '#{escape(col[:comment])}';
EOS
    end
    sql
  end

  def self.flydata_ctl_sql(flydata_tabledef, schema_name)
    flydata_ctl_columns_sql(flydata_tabledef, schema_name) +
    flydata_ctl_tables_sql(flydata_tabledef, schema_name)
  end
  FLYDATA_CTL_COLUMNS_SQL = <<EOS
DELETE FROM %s WHERE table_name = '%s';
INSERT INTO %s (table_name, column_name, src_data_type, ordinal_position) VALUES
EOS
  def self.flydata_ctl_columns_sql(flydata_tabledef, schema_name)
    flydata_ctl_tbl = flydata_ctl_table_for_ddl(schema_name, :columns)
    sql = FLYDATA_CTL_COLUMNS_SQL % [ flydata_ctl_tbl, escape(flydata_tabledef[:table_name]), flydata_ctl_tbl ]
    values = []
    flydata_tabledef[:columns].each.with_index(1) do |col, i|
      charset = col[:charset] ? " cs:#{col[:charset]}" : ""
      values << "('#{escape(flydata_tabledef[:table_name])}', '#{escape(col[:column])}', '#{escape(col[:type])}#{charset}', #{i})"
    end
    sql += values.join(",\n") + ";\n"
    sql
  end

  FLYDATA_CTL_TABLES_SQL = <<EOS
DELETE FROM %s WHERE table_name = '%s';
INSERT INTO %s (table_name, attribute, value) VALUES
EOS

  def self.flydata_ctl_tables_sql(flydata_tabledef, schema_name)
    flydata_ctl_tbl = flydata_ctl_table_for_ddl(schema_name, :tables)
    sql = FLYDATA_CTL_TABLES_SQL % [ flydata_ctl_tbl, flydata_tabledef[:table_name], flydata_ctl_tbl ]
    values = []
    values << "('#{flydata_tabledef[:table_name]}', 'cs', '#{escape(flydata_tabledef[:default_charset])}')"
    values << "('#{flydata_tabledef[:table_name]}', 'revision', 1)"
    values << "('#{flydata_tabledef[:table_name]}', 'src_ddl', '#{escape(flydata_tabledef[:src_ddl])}')"
    if flydata_tabledef[:pk_override]
      values << "('#{flydata_tabledef[:table_name]}', 'pk_override', '#{escape(flydata_tabledef[:pk_override].join(","))}')"
    end
    sql += values.join(",\n") + ";\n"
    sql
  end

  # convert key to valid a column name
  #   http://docs.aws.amazon.com/redshift/latest/dg/r_names.html
  #   http://www.postgresql.org/docs/8.0/static/ddl-system-columns.html
  def self.convert_to_valid_column_name(key)
    convert_to_valid_name(key) {|n| redshift_reserved_column_name?(n) }
  end

  def self.convert_to_valid_table_name(key)
    convert_to_valid_name(key)
  end

  MAX_COLUMNNAME_LENGTH = 127
  VALID_IDENTIFIER_CHARACTERS = [" "] + %w"a-z 0-9 ! # $ % & ' ( ) * + , . \\- \\/ : ; < = > ? @ \\[ \\\\ \\] ^ { \\| } ~ `"

  def self.convert_to_valid_name(key)
    match = /[^#{self::VALID_IDENTIFIER_CHARACTERS.join}]/
    name = key.downcase.gsub(match, '_')
    do_convert = yield name if block_given?
    name = "_#{name}" if do_convert
    if name.length > MAX_COLUMNNAME_LENGTH
      raise NameTooLongError, "Converted name is too long. size: #{name.size} name: #{name}"
    end
    name
  end

  REDSHIFT_RESERVED_COLUMNS = %w[oid tableoid xmin cmin xmax cmax ctid deletexid insertxid]
  REDSHIFT_RESERVED_COLUMNS_HASH = REDSHIFT_RESERVED_COLUMNS.inject({}) {|h, word| h[word.to_sym] = true; h}

  def self.redshift_reserved_column_name?(name)
    return false unless name
    REDSHIFT_RESERVED_COLUMNS_HASH[name.to_sym] == true
  end

  def self.escape(text)
    text.gsub(/(['\\])/, "\\\\\\1")
  end

  def self.check_and_replace_max(params, max_size_a, flydata_type)
      final_params = []
      case flydata_type
      when 'numeric', 'numeric unsigned'
        p = params.split(",")
        raise "Wrong number of params for #{flydata_type} type." if p.length != max_size_a.length
        above_decimal = p[0].to_i - p[1].to_i
        below_decimal = p[1].to_i

        # use max precision as the max number of above-dicimal digits
        max_above_decimal = max_size_a[0]
        final_above_decimal = above_decimal > max_above_decimal ? max_above_decimal : above_decimal

        max_below_decimal = max_size_a[0] - final_above_decimal
        max_below_decimal = max_size_a[1] if max_below_decimal > max_size_a[1]
        final_below_decimal = below_decimal > max_below_decimal ? max_below_decimal : below_decimal

        final_params << (final_above_decimal + final_below_decimal) << final_below_decimal
      else
        params.split(",").each_with_index do |param, i|
          final_params << (/\d+/.match(param) && max_size_a[i] && param.to_i > max_size_a[i].to_i ?
                              max_size_a[i] : param)
        end
      end
      final_params.join(",")
  end

  APACHE_TIMESTAMP_REGEXP = Regexp.new('^(?<apache_time_format>\[[0-3]\d\/\D{3}\/[1-2]\d{3}:[0-2]\d:[0-5]\d:[0-5]\d ?[\+\-]\d{2}:?\d{2}\])$')
  TIME_REGEXP = Regexp.new('^(?<sign>-)?(?<hour>\d{2,3}):(?<minute>[0-5][0-9]):(?<second>[0-5][0-9](\.\d+)?)\s*(?<tz>[+-][0-2][0-9](:?\d{2})?)?$')

  def self.parse_timestamp(value)
    value_str = value.to_s
    return nil if value_str.empty?
    if value.kind_of?(Integer) or /^\d+$/ === value_str
      # Unix epoch in UTC
      t = DateTime.strptime(value_str, '%s')
    elsif APACHE_TIMESTAMP_REGEXP.match(value_str)
      # apache time format
      t = DateTime.strptime(value, "[%d/%b/%Y:%H:%M:%S %Z]")
    elsif time_match = TIME_REGEXP.match(value_str)
      if time_match[:tz]
        # let DateTime.parse handle timezone
        t = DateTime.parse("0001-01-01 #{value_str}")
      else
        # use own converter which takes care of negative hours
        t = convert_time_into_timestamp(time_match)
      end
    elsif /^(\d+)\.(\d+)$/ === value_str # epoch with fraction
      epoch = $1
      fraction = $2
      t = DateTime.strptime(epoch, '%s')
      t = DateTime.parse(t.strftime("%Y-%m-%dT%H:%M:%S.#{fraction}"))
    else
      t = DateTime.parse(convert_date(value_str))
    end
    t = t.new_offset(0) # Redshift Plug-in uses UTC
    t.strftime('%Y-%m-%d %H:%M:%S.%6N')
  rescue ArgumentError => ae
    # '0000-00-00 00:00:00' is valid for mysql datetime column
    if value_str.start_with?('0000-00-00 00:00:00')
      return '0001-01-01 00:00:00.000000'
    else
      raise ae
    end
  end

  STANDARD_DATETIME = DateTime.new(1, 1, 1)

  def self.convert_time_into_timestamp(time_match)
    sign = time_match[:sign] ? -1 : 1
    STANDARD_DATETIME.dup +
      sign * Rational(time_match[:hour].to_i, 24) +
      sign * Rational(time_match[:minute].to_i, 1440) +
      sign * Rational(time_match[:second].to_f, 86400)
  end

  def self.parse_date(value)
    return nil if value.nil?
    value_str = value.to_s
    dt = Date.parse(convert_date(value_str))
    dt.strftime('%Y-%m-%d')
  rescue ArgumentError => ae
    # '0000-00-00' is valid for mysql date column
    return '0001-01-01' if value_str == '0000-00-00'
    raise ae
  end

  CONVERT_ZERO_DATE_REGEXP = /^0000|-00/
  CONVERT_ZERO_DATE_HASH_TABLE = {"0000" => "0001", "-00" => "-01" }

  def self.convert_date(value)
    value = convert_year_into_date(value)
    # Convert any 00 value to 01
    value.gsub(CONVERT_ZERO_DATE_REGEXP, CONVERT_ZERO_DATE_HASH_TABLE)
  end

  def self.convert_year_into_date(value)
    return value unless /^\d+$/.match(value)
    # Only positive integers are allowed for year type
    val_i = value.to_i

    if val_i == 0  # '0' or '0000'
      '0001-01-01' # '0001-01-01' is the "base" date
    elsif value.length == 4
      "#{value}-01-01"
    elsif value.length == 2
      "#{val_i >= 70 ? "19" : "20"}#{value}-01-01"
    else
      value # Return the value as is
    end
  end
end

end
end