Postgresql upsert

upsertable.rb

module Concerns::Upsertable
  extend ActiveSupport::Concern

  CASTABLE_TYPES = [
    "boolean", "date", "integer", "double precision", "hstore",
    "json", "jsonb", "timestamp without time zone"
  ].freeze

  class_methods do

    def casted_column(colname)
      column = self.columns_hash[colname.to_s]
      if CASTABLE_TYPES.include?(column.sql_type)
        "#{colname}::#{column.sql_type}"
      else
        colname.to_s
      end
    end

    # Formats instantiated ActiveModel objects into SQL keys/values input.
    # @returns [Array] [[key1, key2], "(val1a, val2a), (val1b, val2b)"]
    def items_to_keys_values(items)
      return [nil, nil] if items.blank?
      keys   = items.first.attributes.keys - %w{id}
      values = items.each_with_object([]) {|item, values|
        value_line = item.attributes.each_with_object([]) {|(k,v), line|
          case k
          when "id"
            next
          when "created_at", "updated_at"
            line << "CURRENT_TIMESTAMP"
          else
            v = connection.quote(v, columns_hash[k])
            sql_type = columns_hash[k].sql_type
            line << (CASTABLE_TYPES.include?(sql_type) ? "#{v}::#{sql_type}" : v)
          end
        }
        values << "(#{value_line.join(',')})"
      }.join(',')
      [keys, values]
    end

    def upsert!(items)
      keys, values = items_to_keys_values(items)
      return true if values.blank?
      key_string = keys.join(',')
      casted_keys = keys.collect{|k|casted_column(k)}
      set_string = keys.each_index.each_with_object([]) {|idx,ss|
        next if keys[idx] == "created_at"
        ss << "#{keys[idx]} = ni.#{casted_keys[idx]}"
      }.join(",")

      # We could do guid de-duplication here but I think it makes more sense to
      # do it from the caller at them moment.
      #   WITH raw_new_items (#{key_string}) AS ( VALUES #{values} ),
      #   new_items AS ( SELECT DISTINCT ON (guid) * FROM raw_new_items ),
      self.connection.execute <<-EOF
        WITH new_items (#{key_string}) AS ( VALUES #{values} ),
        upsert AS (
          UPDATE #{table_name} all_items
          SET #{set_string}
          FROM new_items ni
          WHERE ni.guid = all_items.guid
          RETURNING all_items.*
        )
        INSERT INTO #{table_name} (#{key_string})
        SELECT #{casted_keys.join(',')}
        FROM new_items
        WHERE NOT EXISTS (SELECT 1
          FROM upsert up
          WHERE up.guid = new_items.guid
        )
      EOF
    end
  end

end