More WIP

Author: robcole

spec/avram/operations/save_operation_spec.cr

@@ -310,16 +310,15 @@ describe "Avram::SaveOperation" do
   describe ".bulk_upsert" do
     context "when the records are not persisted" do
       it "should create the records" do
-        record_args = (1..50).to_a.map do |i|
+        record_args = (1..2).to_a.map do |i|
           {
             :name     => "Test #{i}",
             :nickname => "Test Nickname #{i}",
           }.as(Avram::BulkUpsert::Params)
         end
+        records = UserFactory.new.build_pair
 
-        result = SaveUser.bulk_upsert(record_args)
-
-        pp result
+        SaveUser.bulk_upsert(record_args)
       end
     end
 

src/avram/bulk_upsert.cr

@@ -1,30 +1,57 @@
-class Avram::BulkUpsert
+class Avram::BulkUpsert(T)
   alias Params = Hash(Symbol, String) | Hash(Symbol, String?) | Hash(Symbol, Nil)
 
-  def initialize(@table : TableName,
-                 @records : Array(Params),
-                 @column_names : Array(Symbol) = [] of Symbol)
+  def initialize(@records : Array(Params))
   end
 
   def statement
-    "insert into #{@table}(#{fields}) values(#{values}) returning *"
+    [
+      "insert into #{table}(#{fields})",
+      "values #{value_placeholders}",
+      "ON CONFLICT DO UPDATE SET #{updates}",
+      "returning #{returning}",
+    ].join(" ")
+  end
+
+  private def table
+    T.table_name
+  end
+
+  private def updates
+    conflict_updates = T.column_names.uniq.map do |column|
+      "SET #{column}=EXCLUDED.#{column}"
+    end
+
+    if T.column_names.includes?(:updated_at)
+      conflict_updates.push("SET updated_at=NOW()").join(", ")
+    else
+      conflict_updates.join(", ")
+    end
+  end
+
+  private def returning
+    "id"
   end
 
   private def fields
-    @column_names.join(", ")
+    T.column_names.join(", ")
+  end
+
+  def args
+    @records.map &.values
   end
 
-  private def record_values(record)
+  private def placeholder_values(record)
     values = record.values.map_with_index(1) do |_value, index|
       "$#{index}"
     end.join(", ")
 
     "(#{values})"
   end
 
-  private def values
+  private def value_placeholders
     @records.map do |record|
-      record_values(record)
+      placeholder_values(record)
     end.join(", ")
   end
 end

src/avram/upsert.cr

@@ -101,10 +101,20 @@ module Avram::Upsert
       end
     {% end %}
 
-    def self.bulk_upsert(params : Array(Avram::BulkUpsert::Params))
-      Avram::BulkUpsert
-        .new(T.table_name, params, T.column_names)
-        .statement
+    def self.bulk_upsert(upserts : Array(Avram::BulkUpsert::Params))
+      upsert_keys = upserts.map(&.keys).uniq
+
+      if upsert_keys.size > 1
+        raise "All hashes passed to bulk_upsert must have the same keys."
+      elsif upsert_keys.flatten.any? { |key| !T.column_names.includes?(key) }
+        raise "All keys in hashes must be column names in the table."
+      end
+
+      upsert = Avram::BulkUpsert(T).new(upserts)
+      pp upsert.statement
+      pp upsert.args
+
+      T.database.query(upsert.statement, args: upsert.args)
     end
   end
 end