18192685574

Committed 02 Oct 2025 12:17PM UTC coverage: 99.722% (-0.3%) from 100.0%

Build # 18192685574

Build Type

push

github

Committed by

web-flow

Commit Message

Validate data types of fields upon receive (#94)

Coverage Stats

95 of 98 new or added lines in 10 files covered. (96.94%)

1075 of 1078 relevant lines covered (99.72%)

161.01 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.81

/lib/table_sync/receiving/handler.rb

# frozen_string_literal: true

class TableSync::Receiving::Handler < Rabbit::EventHandler
  extend TableSync::Receiving::DSL

  attr_accessor :version
  attr_reader :event, :model

  def initialize(message)
    super

    self.event   = message.data[:event]
    self.model   = message.data[:model]
    self.version = message.data[:version]
  end

  def call
    configs.each do |config|
      next unless config.allow_event?(event)

      data = processed_data(config)

      next if data.empty?

      version_key = config.version_key(data:)
      data.each { |row| row[version_key] = version }

      target_keys = config.target_keys(data:)

      validate_data(data, target_keys:)

      data.sort_by! { |row| row.values_at(*target_keys).map { |value| sort_key(value) } }

      params = { data:, target_keys:, version_key: }

      if event == :update
        params[:default_values] = config.default_values(data:)
      end

      config.wrap_receiving(event:, **params) do
        perform(config, params)
      end
    end
  end

  private

  # redefine setter from Rabbit::EventHandler
  def data=(raw_data)
    super(Array.wrap(raw_data[:attributes]))
  end

  def event=(event_name)
    event_name = event_name.to_sym

    if event_name.in?(TableSync::Event::VALID_RESOLVED_EVENTS)
      @event = event_name
    else
      raise TableSync::UndefinedEvent.new(event)
    end
  end

  def model=(model_name)
    @model = model_name.to_s
  end

  def configs
    @configs ||= begin
      configs = self.class.configs[model]
      configs = configs.sort_by { |config| "#{config.model.schema}.#{config.model.table}" }
      configs.map do |config|
        ::TableSync::Receiving::ConfigDecorator.new(
          config:,
          # next parameters will be send to each proc-options from config
          event:,
          model:,
          version:,
          project_id:,
          raw_data: data,
        )
      end
    end
  end

  def processed_data(config)
    data.filter_map do |row|
      next if config.skip(row:)

      row = row.dup

      config.mapping_overrides(row:).each do |before, after|
        row[after] = row.delete(before)
      end

      config.except(row:).each { |x| row.delete(x) }

      row.merge!(config.additional_data(row:))

      only = config.only(row:)
      row, rest = row.partition { |key, _| key.in?(only) }.map(&:to_h)

      rest_key = config.rest_key(row:, rest:)
      (row[rest_key] ||= {}).merge!(rest) if rest_key

      row
    end
  end

  def validate_data(data, target_keys:)
    data.each do |row|
      next if target_keys.all? { |x| row.key?(x) }

      raise TableSync::DataError.new(
        data, target_keys, "Some target keys not found in received attributes!"
      )
    end

    if data.uniq { |row| row.slice(*target_keys) }.size != data.size
      raise TableSync::DataError.new(data, target_keys, "Duplicate rows found!")
    end

    keys_sample = data[0].keys
    keys_diff = data.each_with_object(Set.new) do |row, set|
      ((row.keys - keys_sample) | (keys_sample - row.keys)).each { |x| set.add(x) }
    end

    unless keys_diff.empty?
      raise TableSync::DataError.new(data, target_keys, <<~MESSAGE)
        Bad batch structure, check keys: #{keys_diff.to_a}
      MESSAGE
    end
  end

  def validate_data_types(model, data)
    errors = model.validate_types(data)
    return if errors.nil?

    raise TableSync::DataError.new(data, errors.keys, errors.to_json)
  end

  def perform(config, params)
    model = config.model

    model.transaction do
      results = if event == :update
                  config.before_update(**params)
                  validate_data_types(model, params[:data])
                  model.upsert(**params)
                else
                  config.before_destroy(**params)
                  model.destroy(**params)
                end

      model.after_commit do
        TableSync::Instrument.notify table: model.table, schema: model.schema,
                                     count: results.count, event:, direction: :receive
      end

      if event == :update
        model.after_commit { config.after_commit_on_update(**params, results:) }
      else
        model.after_commit { config.after_commit_on_destroy(**params, results:) }
      end
    end
  end

  def sort_key(value)
    value.is_a?(Comparable) ? value : value.to_s
  end
end

1	# frozen_string_literal: true
2
3	class TableSync::Receiving::Handler < Rabbit::EventHandler	4✔
4	extend TableSync::Receiving::DSL	4✔
5
6	attr_accessor :version	4✔
7	attr_reader :event, :model	4✔
8
9	def initialize(message)	4✔
10	super	208✔
11
12	self.event = message.data[:event]	208✔
13	self.model = message.data[:model]	204✔
14	self.version = message.data[:version]	204✔
15	end
16
17	def call	4✔
18	configs.each do \|config\|	200✔
19	next unless config.allow_event?(event)	224✔
20
21	data = processed_data(config)	224✔
22
23	next if data.empty?	224✔
24
25	version_key = config.version_key(data:)	216✔
26	data.each { \|row\| row[version_key] = version }	488✔
27
28	target_keys = config.target_keys(data:)	216✔
29
30	validate_data(data, target_keys:)	216✔
31
32	data.sort_by! { \|row\| row.values_at(*target_keys).map { \|value\| sort_key(value) } }	716✔
33
34	params = { data:, target_keys:, version_key: }	204✔
35
36	if event == :update	204✔
37	params[:default_values] = config.default_values(data:)	140✔
38	end
39
40	config.wrap_receiving(event:, **params) do	204✔
41	perform(config, params)	204✔
42	end
43	end
44	end
45
46	private	4✔
47
48	# redefine setter from Rabbit::EventHandler
49	def data=(raw_data)	4✔
50	super(Array.wrap(raw_data[:attributes]))	208✔
51	end
52
53	def event=(event_name)	4✔
54	event_name = event_name.to_sym	208✔
55
56	if event_name.in?(TableSync::Event::VALID_RESOLVED_EVENTS)	208✔
57	@event = event_name	204✔
58	else
59	raise TableSync::UndefinedEvent.new(event)	4✔
60	end
61	end
62
63	def model=(model_name)	4✔
64	@model = model_name.to_s	204✔
65	end
66
67	def configs	4✔
68	@configs \|\|= begin	200✔
69	configs = self.class.configs[model]	200✔
70	configs = configs.sort_by { \|config\| "#{config.model.schema}.#{config.model.table}" }	424✔
71	configs.map do \|config\|	200✔
72	::TableSync::Receiving::ConfigDecorator.new(	224✔
73	config:,
74	# next parameters will be send to each proc-options from config
75	event:,
76	model:,
77	version:,
78	project_id:,
79	raw_data: data,
80	)
81	end
82	end
83	end
84
85	def processed_data(config)	4✔
86	data.filter_map do \|row\|	224✔
87	next if config.skip(row:)	280✔
88
89	row = row.dup	272✔
90
91	config.mapping_overrides(row:).each do \|before, after\|	272✔
92	row[after] = row.delete(before)	88✔
93	end
94
95	config.except(row:).each { \|x\| row.delete(x) }	280✔
96
97	row.merge!(config.additional_data(row:))	272✔
98
99	only = config.only(row:)	272✔
100	row, rest = row.partition { \|key, _\| key.in?(only) }.map(&:to_h)	1,240✔
101
102	rest_key = config.rest_key(row:, rest:)	272✔
103	(row[rest_key] \|\|= {}).merge!(rest) if rest_key	272✔
104
105	row	272✔
106	end
107	end
108
109	def validate_data(data, target_keys:)	4✔
110	data.each do \|row\|	216✔
111	next if target_keys.all? { \|x\| row.key?(x) }	552✔
112
113	raise TableSync::DataError.new(	4✔
114	data, target_keys, "Some target keys not found in received attributes!"
115	)
116	end
117
118	if data.uniq { \|row\| row.slice(*target_keys) }.size != data.size	316✔
119	raise TableSync::DataError.new(data, target_keys, "Duplicate rows found!")	4✔
120	end
121
122	keys_sample = data[0].keys	208✔
123	keys_diff = data.each_with_object(Set.new) do \|row, set\|	208✔
124	((row.keys - keys_sample) \| (keys_sample - row.keys)).each { \|x\| set.add(x) }	260✔
125	end
126
127	unless keys_diff.empty?	208✔
128	raise TableSync::DataError.new(data, target_keys, <<~MESSAGE)	4✔
129	Bad batch structure, check keys: #{keys_diff.to_a}
130	MESSAGE
131	end
132	end
133
134	def validate_data_types(model, data)	4✔
135	errors = model.validate_types(data)	140✔
136	return if errors.nil?	140✔
137
NEW 138	raise TableSync::DataError.new(data, errors.keys, errors.to_json)	×
139	end
140
141	def perform(config, params)	4✔
142	model = config.model	204✔
143
144	model.transaction do	204✔
145	results = if event == :update	204✔
146	config.before_update(**params)	140✔
147	validate_data_types(model, params[:data])	140✔
148	model.upsert(**params)	140✔
149	else
150	config.before_destroy(**params)	64✔
151	model.destroy(**params)	64✔
152	end
153
154	model.after_commit do	204✔
155	TableSync::Instrument.notify table: model.table, schema: model.schema,	204✔
156	count: results.count, event:, direction: :receive
157	end
158
159	if event == :update	204✔
160	model.after_commit { config.after_commit_on_update(**params, results:) }	280✔
161	else
162	model.after_commit { config.after_commit_on_destroy(**params, results:) }	128✔
163	end
164	end
165	end
166
167	def sort_key(value)	4✔
168	value.is_a?(Comparable) ? value : value.to_s	264✔
169	end
170	end

umbrellio / table_sync / 18192685574

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous