Add new option.

Author: omeH

lib/table_sync/receiving.rb

@@ -6,6 +6,7 @@ module Receiving
     require_relative "receiving/config_decorator"
     require_relative "receiving/dsl"
     require_relative "receiving/handler"
+    require_relative "receiving/hooks/once"
     require_relative "receiving/model/active_record"
     require_relative "receiving/model/sequel"
   end

lib/table_sync/receiving/config.rb

@@ -218,7 +218,7 @@ def option(name)
   value_as_proc_setter_wrapper: any_value,
   default: proc { proc { |&block| block.call } }
 
-TableSync::Receiving::Config.add_simple_option :conditional_handler
+TableSync::Receiving::Config.add_simple_option :on_first_sync
 
 %i[
   before_update

lib/table_sync/receiving/handler.rb

@@ -160,6 +160,12 @@ def perform(config, params) # rubocop:disable Metrics/MethodLength
       if event == :update
         model.after_commit do
           config.option(:after_commit_on_update, **params, results:)
+
+          conditions, handler = config.option(:on_first_sync)
+          if conditions.present?
+            hook = ::TableSync::Receiving::Hooks::Once.new(conditions:, config:)
+            hook.perform(targets: results, &handler)
+          end
         end
       else
         model.after_commit { config.option(:after_commit_on_destroy, **params, results:) }

lib/table_sync/receiving/hooks/once.rb

@@ -0,0 +1,49 @@
+# frozen_string_literal: true
+
+module TableSync::Receiving::Hooks
+  class Once
+    attr_reader :conditions, :config
+
+    def initialize(conditions:, config:)
+      @conditions = conditions
+      @config = config
+    end
+
+    def perform(targets:, &)
+      target_keys = config.option(:target_keys)
+      model = config.model
+
+      targets.each do |target|
+        next unless conditions?(target)
+
+        model.transaction(isolation: model.isolation_level(:repeatable)) do
+          model.find_and_update(row: target, target_keys:) do |entry|
+            next unless allow?(entry)
+
+            entry.hooks ||= []
+            entry.hooks << hook_lookup_code
+            model.after_commit { yield(entry:) }
+          end
+        end
+      end
+    end
+
+    private
+
+    def allow?(entry)
+      Array(entry.hooks).exclude?(hook_lookup_code)
+    end
+
+    def hook_lookup_code
+      @hook_lookup_code ||= conditions[:columns].map do |column|
+        "#{column}-#{conditions[column]}"
+      end.join(":")
+    end
+
+    def conditions?(row)
+      conditions[:columns].all? do |column|
+        row[column] == (conditions[column] || row[column])
+      end
+    end
+  end
+end

lib/table_sync/receiving/model/active_record.rb

@@ -2,6 +2,14 @@
 
 module TableSync::Receiving::Model
   class ActiveRecord
+    ISOLATION_LEVELS = {
+      uncommitted: :read_uncommitted,
+      committed: :read_committed,
+      repeatable: :repeatable_read,
+      serializable: :serializable,
+    }.freeze
+    private_constant :ISOLATION_LEVELS
+
     class AfterCommitWrap
       def initialize(&block)
         @callback = block
@@ -33,6 +41,10 @@ def initialize(table_name)
       @schema = model_naming.schema.to_sym
     end
 
+    def isolation_level(lookup_code)
+      ISOLATION_LEVELS.fetch(lookup_code)
+    end
+
     def columns
       raw_model.column_names.map(&:to_sym)
     end
@@ -110,14 +122,20 @@ def validate_types(data)
       types_validator.validate(data)
     end
 
-    def transaction(&)
-      ::ActiveRecord::Base.transaction(&)
+    def transaction(**params, &)
+      ::ActiveRecord::Base.transaction(**params, &)
     end
 
     def after_commit(&)
       db.add_transaction_record(AfterCommitWrap.new(&))
     end
 
+    def find_and_update(row:, target_keys:)
+      entry = raw_model.find_by!(row.slice(*target_keys))
+      yield entry
+      entry.save!
+    end
+
     private
 
     attr_reader :raw_model, :types_validator

lib/table_sync/receiving/model/sequel.rb

@@ -4,6 +4,14 @@ module TableSync::Receiving::Model
   class Sequel
     attr_reader :table, :schema
 
+    ISOLATION_LEVELS = {
+      uncommitted: :uncommitted,
+      committed: :committed,
+      repeatable: :repeatable,
+      serializable: :serializable,
+    }.freeze
+    private_constant :ISOLATION_LEVELS
+
     def initialize(table_name)
       @raw_model = Class.new(::Sequel::Model(table_name)).tap(&:unrestrict_primary_key)
       @types_validator = TableSync::Utils::Schema::Builder::Sequel.build(@raw_model)
@@ -17,6 +25,10 @@ def initialize(table_name)
       @schema = model_naming.schema.to_sym
     end
 
+    def isolation_level(lookup_code)
+      ISOLATION_LEVELS.fetch(lookup_code)
+    end
+
     def columns
       dataset.columns
     end
@@ -57,14 +69,20 @@ def validate_types(data)
       types_validator.validate(data)
     end
 
-    def transaction(&)
-      db.transaction(&)
+    def transaction(**params, &)
+      db.transaction(**params, &)
     end
 
     def after_commit(&)
       db.after_commit(&)
     end
 
+    def find_and_update(row:, target_keys:)
+      entry = dataset.first!(row.slice(*target_keys))
+      yield entry
+      entry.save_changes
+    end
+
     private
 
     attr_reader :raw_model, :types_validator

spec/receiving/config_spec.rb

@@ -349,33 +349,33 @@
       end
     end
 
-    describe "#conditional_handler" do
+    describe "#on_first_sync" do
       it "returns correct default value" do
-        value, callable = config.option(:conditional_handler)
+        value, callable = config.option(:on_first_sync)
         expect(value).to be_nil
         expect(callable).to be_a(Proc)
       end
 
       it "processes a single value" do
-        config.conditional_handler :test
+        config.on_first_sync :test
 
-        value, callable = config.option(:conditional_handler)
+        value, callable = config.option(:on_first_sync)
         expect(value).to eq(:test)
         expect(callable).to be_a(Proc)
       end
 
       it "processes a single block" do
-        config.conditional_handler { :test }
+        config.on_first_sync { :test }
 
-        value, callable = config.option(:conditional_handler)
+        value, callable = config.option(:on_first_sync)
         expect(value).to be_nil
         expect(callable.call).to eq(:test)
       end
 
       it "processes a value and a block" do
-        config.conditional_handler(:spam) { :test }
+        config.on_first_sync(:spam) { :test }
 
-        value, callable = config.option(:conditional_handler)
+        value, callable = config.option(:on_first_sync)
         expect(value).to eq(:spam)
         expect(callable.call).to eq(:test)
       end

spec/receiving/handler_spec.rb

@@ -171,6 +171,7 @@ def fire_destroy_event
         online_status: nil,
         project_id: "PID",
         rest: nil,
+        hooks: nil,
         version: 123.34534,
       }
     end
@@ -362,6 +363,7 @@ def fire_destroy_event
             online_status: nil,
             project_id: "PID",
             rest: nil,
+            hooks: nil,
             version: 123.34534,
           )
 
@@ -371,6 +373,7 @@ def fire_destroy_event
             online_status: nil,
             project_id: "PID",
             rest: nil,
+            hooks: nil,
             version: 123.34534,
           )
         end
@@ -605,6 +608,51 @@ def destroy(data:, target_keys:, version_key:)
     end
   end
 
+  describe "#on_first_sync" do
+    let(:callback_flags) { { on_first_sync: [] } }
+    let(:handler) do
+      callback_flags_link = callback_flags
+      handler = Class.new(described_class)
+
+      handler.receive("User", to_table: :players) do
+        rest_key false
+        mapping_overrides id: :external_id
+
+        on_first_sync columns: %i[online_status], online_status: true do |entry:|
+          callback_flags_link[:on_first_sync] << entry.external_id
+        end
+      end
+
+      handler
+    end
+    let(:update_event) do
+      OpenStruct.new(
+        data: {
+          event: "update",
+          model: "User",
+          attributes: {
+            id: user_id,
+            email: "[email protected]",
+            online_status: true,
+          },
+          version: 123.34534,
+        },
+        project_id: "pid",
+      )
+    end
+
+    it "provides proper event to wrap receiving" do
+      fire_update_event
+      expect(DB[:players].count).to eq(1)
+      expect(callback_flags[:on_first_sync]).to be_one
+
+      update_event.data[:version] = update_event.data[:version] + 1
+      fire_update_event
+      expect(DB[:players].count).to eq(1)
+      expect(callback_flags[:on_first_sync]).to be_one
+    end
+  end
+
   describe "avoid dead locks" do
     let(:model) do
       Class.new(TableSync.receiving_model) do

spec/receiving/models/active_record_spec.rb

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+describe TableSync::Receiving::Model::ActiveRecord do
+  subject(:model) { described_class.new(:players) }
+
+  describe "#isolation_level" do
+    it { expect(model.isolation_level(:uncommitted)).to eq(:read_uncommitted) }
+    it { expect(model.isolation_level(:committed)).to eq(:read_committed) }
+    it { expect(model.isolation_level(:repeatable)).to eq(:repeatable_read) }
+    it { expect(model.isolation_level(:serializable)).to eq(:serializable) }
+    it { expect { model.isolation_level(:invalid) }.to raise_error(KeyError) }
+  end
+
+  describe "find_and_update" do
+    let(:primary_key) { :external_id }
+    let(:external_id) { 100_500 }
+    let!(:player) do
+      model.send(:raw_model).create(
+        external_id:,
+        email: "[email protected]",
+        online_status: false,
+        version: 123.456,
+      )
+    end
+
+    let(:row) { { external_id: } }
+
+    it "finds and updates an entry" do
+      model.find_and_update(row:, target_keys: [primary_key]) do |entry|
+        entry.online_status = true
+      end
+      expect(player.reload.online_status).to be_truthy
+    end
+
+    it "raise an error" do
+      expect do
+        row = { external_id: external_id + 1 }
+        model.find_and_update(row:, target_keys: [primary_key]) do |entry|
+          entry.online_status = true
+        end
+      end.to raise_error(ActiveRecord::RecordNotFound)
+      expect(player.reload.online_status).to be_falsy
+    end
+  end
+end

spec/receiving/models/sequel_spec.rb

@@ -0,0 +1,45 @@
+# frozen_string_literal: true
+
+describe TableSync::Receiving::Model::Sequel do
+  subject(:model) { described_class.new(:players) }
+
+  describe "#isolation_level" do
+    it { expect(model.isolation_level(:uncommitted)).to eq(:uncommitted) }
+    it { expect(model.isolation_level(:committed)).to eq(:committed) }
+    it { expect(model.isolation_level(:repeatable)).to eq(:repeatable) }
+    it { expect(model.isolation_level(:serializable)).to eq(:serializable) }
+    it { expect { model.isolation_level(:invalid) }.to raise_error(KeyError) }
+  end
+
+  describe "find_and_update" do
+    let(:primary_key) { :external_id }
+    let(:external_id) { 100_500 }
+    let!(:player) do
+      model.send(:raw_model).create(
+        external_id:,
+        email: "[email protected]",
+        online_status: false,
+        version: 123.456,
+      )
+    end
+
+    let(:row) { { external_id: } }
+
+    it "finds and updates an entry" do
+      model.find_and_update(row:, target_keys: [primary_key]) do |entry|
+        entry.online_status = true
+      end
+      expect(player.reload.online_status).to be_truthy
+    end
+
+    it "raise an error" do
+      expect do
+        row = { external_id: external_id + 1 }
+        model.find_and_update(row:, target_keys: [primary_key]) do |entry|
+          entry.online_status = true
+        end
+      end.to raise_error(Sequel::NoMatchingRow)
+      expect(player.reload.online_status).to be_falsy
+    end
+  end
+end