class ActiveRecord::Base

def self.synchronize(instances, keys = [primary_key])
  return if instances.empty?
  conditions = {}
  key_values = keys.map { |key| instances.map(&key.to_sym) }
  keys.zip(key_values).each { |key, values| conditions[key] = values }
  order = keys.map { |key| "#{key} ASC" }.join(",")
  klass = instances.first.class
  fresh_instances = klass.where(conditions).order(order)
  instances.each do |instance|
    matched_instance = fresh_instances.detect do |fresh_instance|
      keys.all? { |key| fresh_instance.send(key) == instance.send(key) }
    end
    next unless matched_instance
    instance.send :clear_aggregation_cache
    instance.send :clear_association_cache
    instance.instance_variable_set :@attributes, matched_instance.instance_variable_get(:@attributes)
    if instance.respond_to?(:clear_changes_information)
      instance.clear_changes_information                      # Rails 4.2 and higher
    else
      instance.instance_variable_set :@attributes_cache, {}   # Rails 4.0, 4.1
      instance.changed_attributes.clear                       # Rails 3.2
      instance.previous_changes.clear
    end
    # Since the instance now accurately reflects the record in
    # the database, ensure that instance.persisted? is true.
    instance.instance_variable_set '@new_record', false
    instance.instance_variable_set '@destroyed', false
  end
end

def add_special_rails_stamps( column_names, array_of_attributes, options )
  timestamp_columns = {}
  timestamps        = {}
  if respond_to?(:all_timestamp_attributes_in_model, true) # Rails 5.1 and higher
    timestamp_columns[:create] = timestamp_attributes_for_create_in_model
    timestamp_columns[:update] = timestamp_attributes_for_update_in_model
  else
    instance = new
    timestamp_columns[:create] = instance.send(:timestamp_attributes_for_create_in_model)
    timestamp_columns[:update] = instance.send(:timestamp_attributes_for_update_in_model)
  end
  # use tz as set in ActiveRecord::Base
  timestamp = ActiveRecord::Base.default_timezone == :utc ? Time.now.utc : Time.now
  [:create, :update].each do |action|
    timestamp_columns[action].each do |column|
      column = column.to_s
      timestamps[column] = timestamp
      index = column_names.index(column) || column_names.index(column.to_sym)
      if index
        # replace every instance of the array of attributes with our value
        array_of_attributes.each { |arr| arr[index] = timestamp if arr[index].nil? || action == :update }
      else
        column_names << column
        array_of_attributes.each { |arr| arr << timestamp }
      end
      if supports_on_duplicate_key_update? && action == :update
        connection.add_column_for_on_duplicate_key_update(column, options)
      end
    end
  end
  timestamps
end

def establish_connection_with_activerecord_import(*args)
  conn = establish_connection_without_activerecord_import(*args)
  if !ActiveRecord.const_defined?(:Import) || !ActiveRecord::Import.respond_to?(:load_from_connection_pool)
    require "activerecord-import/base"
  end
  ActiveRecord::Import.load_from_connection_pool connection_pool
  conn
end

def find_associated_objects_for_import(associated_objects_by_class, model)
  associated_objects_by_class[model.class.name] ||= {}
  association_reflections =
    model.class.reflect_on_all_associations(:has_one) +
    model.class.reflect_on_all_associations(:has_many)
  association_reflections.each do |association_reflection|
    associated_objects_by_class[model.class.name][association_reflection.name] ||= []
    association = model.association(association_reflection.name)
    association.loaded!
    # Wrap target in an array if not already
    association = Array(association.target)
    changed_objects = association.select { |a| a.new_record? || a.changed? }
    changed_objects.each do |child|
      child.public_send("#{association_reflection.foreign_key}=", model.id)
      # For polymorphic associations
      association_reflection.type.try do |type|
        child.public_send("#{type}=", model.class.base_class.name)
      end
    end
    associated_objects_by_class[model.class.name][association_reflection.name].concat changed_objects
  end
  associated_objects_by_class
end

def import(*args)
  if args.first.is_a?( Array ) && args.first.first.is_a?(ActiveRecord::Base)
    options = {}
    options.merge!( args.pop ) if args.last.is_a?(Hash)
    models = args.first
    import_helper(models, options)
  else
    import_helper(*args)
  end
end

def import!(*args)
  options = args.last.is_a?( Hash ) ? args.pop : {}
  options[:validate] = true
  options[:raise_error] = true
  import(*args, options)
end

def import_associations(models, options)
  # now, for all the dirty associations, collect them into a new set of models, then recurse.
  # notes:
  #    does not handle associations that reference themselves
  #    should probably take a hash to associations to follow.
  return if models.nil?
  associated_objects_by_class = {}
  models.each { |model| find_associated_objects_for_import(associated_objects_by_class, model) }
  # :on_duplicate_key_update not supported for associations
  options.delete(:on_duplicate_key_update)
  associated_objects_by_class.each_value do |associations|
    associations.each_value do |associated_records|
      associated_records.first.class.import(associated_records, options) unless associated_records.empty?
    end
  end
end

def import_helper( *args )
  options = { validate: true, timestamps: true }
  options.merge!( args.pop ) if args.last.is_a? Hash
  # making sure that current model's primary key is used
  options[:primary_key] = primary_key
  # Don't modify incoming arguments
  if options[:on_duplicate_key_update] && options[:on_duplicate_key_update].duplicable?
    options[:on_duplicate_key_update] = options[:on_duplicate_key_update].dup
  end
  is_validating = options[:validate]
  is_validating = true unless options[:validate_with_context].nil?
  # assume array of model objects
  if args.last.is_a?( Array ) && args.last.first.is_a?(ActiveRecord::Base)
    if args.length == 2
      models = args.last
      column_names = args.first.dup
    else
      models = args.first
      column_names = self.column_names.dup
    end
    if models.first.id.nil? && column_names.include?(primary_key) && columns_hash[primary_key].type == :uuid
      column_names.delete(primary_key)
    end
    default_values = column_defaults
    stored_attrs = respond_to?(:stored_attributes) ? stored_attributes : {}
    serialized_attrs = if defined?(ActiveRecord::Type::Serialized)
      attrs = column_names.select { |c| type_for_attribute(c.to_s).class == ActiveRecord::Type::Serialized }
      Hash[attrs.map { |a| [a, nil] }]
    else
      serialized_attributes
    end
    array_of_attributes = models.map do |model|
      column_names.map do |name|
        if stored_attrs.key?(name.to_sym) ||
           serialized_attrs.key?(name) ||
           default_values.key?(name.to_s)
          model.read_attribute(name.to_s)
        else
          model.read_attribute_before_type_cast(name.to_s)
        end
      end
    end
    # supports array of hash objects
  elsif args.last.is_a?( Array ) && args.last.first.is_a?(Hash)
    if args.length == 2
      array_of_hashes = args.last
      column_names = args.first.dup
    else
      array_of_hashes = args.first
      column_names = array_of_hashes.first.keys
    end
    array_of_attributes = array_of_hashes.map do |h|
      column_names.map do |key|
        h[key]
      end
    end
    # supports empty array
  elsif args.last.is_a?( Array ) && args.last.empty?
    return ActiveRecord::Import::Result.new([], 0, [])
    # supports 2-element array and array
  elsif args.size == 2 && args.first.is_a?( Array ) && args.last.is_a?( Array )
    unless args.last.first.is_a?(Array)
      raise ArgumentError, "Last argument should be a two dimensional array '[[]]'. First element in array was a #{args.last.first.class}"
    end
    column_names, array_of_attributes = args
    # dup the passed args so we don't modify unintentionally
    column_names = column_names.dup
    array_of_attributes = array_of_attributes.map(&:dup)
  else
    raise ArgumentError, "Invalid arguments!"
  end
  # Force the primary key col into the insert if it's not
  # on the list and we are using a sequence and stuff a nil
  # value for it into each row so the sequencer will fire later
  symbolized_column_names = Array(column_names).map(&:to_sym)
  symbolized_primary_key = Array(primary_key).map(&:to_sym)
  if !symbolized_primary_key.to_set.subset?(symbolized_column_names.to_set) && connection.prefetch_primary_key? && sequence_name
    column_count = column_names.size
    column_names.concat(primary_key).uniq!
    columns_added = column_names.size - column_count
    new_fields = Array.new(columns_added)
    array_of_attributes.each { |a| a.concat(new_fields) }
  end
  timestamps = {}
  # record timestamps unless disabled in ActiveRecord::Base
  if record_timestamps && options.delete( :timestamps )
    timestamps = add_special_rails_stamps column_names, array_of_attributes, options
  end
  return_obj = if is_validating
    if models
      import_with_validations( column_names, array_of_attributes, options ) do |validator, failed|
        models.each_with_index do |model, i|
          model = model.dup if options[:recursive]
          next if validator.valid_model? model
          raise(ActiveRecord::RecordInvalid, model) if options[:raise_error]
          array_of_attributes[i] = nil
          failed << model
        end
      end
    else
      import_with_validations( column_names, array_of_attributes, options )
    end
  else
    (num_inserts, ids) = import_without_validations_or_callbacks( column_names, array_of_attributes, options )
    ActiveRecord::Import::Result.new([], num_inserts, ids)
  end
  if options[:synchronize]
    sync_keys = options[:synchronize_keys] || [primary_key]
    synchronize( options[:synchronize], sync_keys)
  end
  return_obj.num_inserts = 0 if return_obj.num_inserts.nil?
  # if we have ids, then set the id on the models and mark the models as clean.
  if models && support_setting_primary_key_of_imported_objects?
    set_attributes_and_mark_clean(models, return_obj, timestamps)
    # if there are auto-save associations on the models we imported that are new, import them as well
    import_associations(models, options.dup) if options[:recursive]
  end
  return_obj
end

def import_with_validations( column_names, array_of_attributes, options = {} )
  failed_instances = []
  validator = ActiveRecord::Import::Validator.new(options)
  if block_given?
    yield validator, failed_instances
  else
    # create instances for each of our column/value sets
    arr = validations_array_for_column_names_and_attributes( column_names, array_of_attributes )
    # keep track of the instance and the position it is currently at. if this fails
    # validation we'll use the index to remove it from the array_of_attributes
    model = new
    arr.each_with_index do |hsh, i|
      hsh.each_pair { |k, v| model[k] = v }
      next if validator.valid_model? model
      raise(ActiveRecord::RecordInvalid, model) if options[:raise_error]
      array_of_attributes[i] = nil
      failure = model.dup
      failure.errors.send(:initialize_dup, model.errors)
      failed_instances << failure
    end
  end
  array_of_attributes.compact!
  num_inserts, ids = if array_of_attributes.empty? || options[:all_or_none] && failed_instances.any?
    [0, []]
  else
    import_without_validations_or_callbacks( column_names, array_of_attributes, options )
  end
  ActiveRecord::Import::Result.new(failed_instances, num_inserts, ids)
end

def import_without_validations_or_callbacks( column_names, array_of_attributes, options = {} )
  column_names = column_names.map(&:to_sym)
  scope_columns, scope_values = scope_attributes.to_a.transpose
  unless scope_columns.blank?
    scope_columns.zip(scope_values).each do |name, value|
      name_as_sym = name.to_sym
      next if column_names.include?(name_as_sym)
      is_sti = (name_as_sym == inheritance_column.to_sym && self < base_class)
      value = value.first if is_sti
      column_names << name_as_sym
      array_of_attributes.each { |attrs| attrs << value }
    end
  end
  columns = column_names.each_with_index.map do |name, i|
    column = columns_hash[name.to_s]
    raise ActiveRecord::Import::MissingColumnError.new(name.to_s, i) if column.nil?
    column
  end
  columns_sql = "(#{column_names.map { |name| connection.quote_column_name(name) }.join(',')})"
  pre_sql_statements = connection.pre_sql_statements( options )
  insert_sql = ['INSERT', pre_sql_statements, "INTO #{quoted_table_name} #{columns_sql} VALUES "]
  insert_sql = insert_sql.flatten.join(' ')
  values_sql = values_sql_for_columns_and_attributes(columns, array_of_attributes)
  number_inserted = 0
  ids = []
  if supports_import?
    # generate the sql
    post_sql_statements = connection.post_sql_statements( quoted_table_name, options )
    batch_size = options[:batch_size] || values_sql.size
    values_sql.each_slice(batch_size) do |batch_values|
      # perform the inserts
      result = connection.insert_many( [insert_sql, post_sql_statements].flatten,
        batch_values,
        options,
        "#{self.class.name} Create Many Without Validations Or Callbacks" )
      number_inserted += result[0]
      ids += result[1]
    end
  else
    transaction(requires_new: true) do
      values_sql.each do |values|
        ids << connection.insert(insert_sql + values)
        number_inserted += 1
      end
    end
  end
  [number_inserted, ids]
end

def set_attributes_and_mark_clean(models, import_result, timestamps)
  return if models.nil?
  models -= import_result.failed_instances
  models.each do |model|
    if model.respond_to?(:clear_changes_information) # Rails 4.0 and higher
      model.clear_changes_information
    else # Rails 3.2
      model.instance_variable_get(:@changed_attributes).clear
    end
    model.instance_variable_set(:@new_record, false)
  end
  # if ids were returned for all models we know all were updated
  if models.size == import_result.ids.size
    import_result.ids.each_with_index do |id, index|
      model = models[index]
      model.id = id
      timestamps.each do |attr, value|
        model.send(attr + "=", value)
      end
    end
  end
end

def support_setting_primary_key_of_imported_objects?
  connection.respond_to?(:support_setting_primary_key_of_imported_objects?) && connection.support_setting_primary_key_of_imported_objects?
end

def supports_import?(*args)
  connection.respond_to?(:supports_import?) && connection.supports_import?(*args)
end

def supports_on_duplicate_key_update?
  connection.supports_on_duplicate_key_update?
end

def synchronize(instances, key = [ActiveRecord::Base.primary_key])
  self.class.synchronize(instances, key)
end

def validations_array_for_column_names_and_attributes( column_names, array_of_attributes ) # :nodoc:
  array_of_attributes.map { |values| Hash[column_names.zip(values)] }
end

def values_sql_for_columns_and_attributes(columns, array_of_attributes) # :nodoc:
  # connection gets called a *lot* in this high intensity loop.
  # Reuse the same one w/in the loop, otherwise it would keep being re-retreived (= lots of time for large imports)
  connection_memo = connection
  array_of_attributes.map do |arr|
    my_values = arr.each_with_index.map do |val, j|
      column = columns[j]
      # be sure to query sequence_name *last*, only if cheaper tests fail, because it's costly
      if val.nil? && column.name == primary_key && !sequence_name.blank?
        connection_memo.next_value_for_sequence(sequence_name)
      elsif column
        if respond_to?(:type_caster)                                         # Rails 5.0 and higher
          type = type_for_attribute(column.name)
          val = type.type == :boolean ? type.cast(val) : type.serialize(val)
          connection_memo.quote(val)
        elsif column.respond_to?(:type_cast_from_user)                       # Rails 4.2
          connection_memo.quote(column.type_cast_from_user(val), column)
        else                                                                 # Rails 3.2, 4.0 and 4.1
          if serialized_attributes.include?(column.name)
            val = serialized_attributes[column.name].dump(val)
          end
          connection_memo.quote(column.type_cast(val), column)
        end
      end
    end
    "(#{my_values.join(',')})"
  end
end