From b8543c09fbb2e7ce2a3ffb34f4c66a963365443d Mon Sep 17 00:00:00 2001 From: Shivam Mishra Date: Fri, 13 Mar 2026 15:17:46 +0530 Subject: [PATCH] refactor: unify backfill service to per-dimension query pattern Replace the two-path aggregation (group-then-fan-out for standard events, per-dimension for distinct-count) with a single per-dimension loop for both. This removes the bug-prone fan-out logic and the conceptual split. Strengthen specs to cover all three dimensions (account, agent, inbox), assert total rollup row counts, and add distinct-count deduplication test. --- .../reporting_events/backfill_service.rb | 125 ++++++++---------- .../reporting_events/backfill_service_spec.rb | 73 ++++++++-- 2 files changed, 122 insertions(+), 76 deletions(-) diff --git a/app/services/reporting_events/backfill_service.rb b/app/services/reporting_events/backfill_service.rb index 8ec1c242e..0deef624d 100644 --- a/app/services/reporting_events/backfill_service.rb +++ b/app/services/reporting_events/backfill_service.rb @@ -1,29 +1,20 @@ # frozen_string_literal: true class ReportingEvents::BackfillService - AGGREGATE_SELECTS = [ - :name, - :user_id, - :inbox_id, - Arel.sql('COUNT(*)'), - Arel.sql('COALESCE(SUM(value), 0)'), - Arel.sql('COALESCE(SUM(value_in_business_hours), 0)') - ].freeze - - DISTINCT_AGGREGATE_SELECTS = [ - :name, - :user_id, - :inbox_id, - Arel.sql('COUNT(DISTINCT conversation_id)'), - Arel.sql('COALESCE(SUM(value), 0)'), - Arel.sql('COALESCE(SUM(value_in_business_hours), 0)') + DIMENSIONS = [ + { type: 'account', group_column: nil }, + { type: 'agent', group_column: :user_id }, + { type: 'inbox', group_column: :inbox_id } ].freeze # TODO: Move this to EventMetricRegistry when we expand distinct-counting support. # The live path already guards uniqueness in ReportingEventListener#conversation_bot_handoff, # but historical duplicates can exist since it's not enforced at the DB level. + # These events are queried per-dimension (not group-then-sum) because COUNT(DISTINCT) is not additive. DISTINCT_COUNT_EVENTS = %w[conversation_bot_handoff].freeze + DISTINCT_COUNT_SQL = Arel.sql('COUNT(DISTINCT conversation_id)') + def self.backfill_date(account, date) new(account, date).perform end @@ -74,72 +65,70 @@ class ReportingEvents::BackfillService def build_aggregates(start_utc, end_utc) aggregates = Hash.new { |h, k| h[k] = { count: 0, sum_value: 0.0, sum_value_business_hours: 0.0 } } + standard_names = ReportingEvents::EventMetricRegistry.event_names - DISTINCT_COUNT_EVENTS + base = @account.reporting_events.where(created_at: start_utc...end_utc) - grouped_events(start_utc, end_utc).each { |grouped_event| accumulate_grouped_aggregates(aggregates, grouped_event) } + DIMENSIONS.each do |dimension| + aggregate_standard_events(aggregates, base.where(name: standard_names), dimension) + aggregate_distinct_events(aggregates, base.where(name: DISTINCT_COUNT_EVENTS), dimension) + end aggregates end - def grouped_events(start_utc, end_utc) - standard = fetch_grouped_events(start_utc, end_utc, standard_event_names, AGGREGATE_SELECTS) - distinct = fetch_grouped_events(start_utc, end_utc, DISTINCT_COUNT_EVENTS, DISTINCT_AGGREGATE_SELECTS) + def aggregate_standard_events(aggregates, scope, dimension) + group_cols, selects = dimension_groups_and_selects(dimension) - (standard + distinct).map { |grouped_row| grouped_event_attributes(grouped_row) } - end + scope.group(*group_cols).pluck(*selects).each do |row| + event_name, dimension_id, count, sum_value, sum_value_business_hours = unpack_row(row, dimension) + next if dimension_id.nil? - def standard_event_names - ReportingEvents::EventMetricRegistry.event_names - DISTINCT_COUNT_EVENTS - end - - def fetch_grouped_events(start_utc, end_utc, event_names, selects) - return [] if event_names.empty? - - @account.reporting_events - .where(name: event_names, created_at: start_utc...end_utc) - .group(:name, :user_id, :inbox_id) - .pluck(*selects) - end - - def dimensions(grouped_event) - { - 'account' => @account.id, - 'agent' => grouped_event[:user_id], - 'inbox' => grouped_event[:inbox_id] - } - end - - def accumulate_grouped_aggregates(aggregates, grouped_event) - ReportingEvents::EventMetricRegistry.metrics_for_aggregate( - grouped_event[:event_name], - count: grouped_event[:count], - sum_value: grouped_event[:sum_value], - sum_value_business_hours: grouped_event[:sum_value_business_hours] - ).each do |metric, metric_data| - accumulate_metric_aggregates(aggregates, dimensions(grouped_event), metric, metric_data) + ReportingEvents::EventMetricRegistry.metrics_for_aggregate( + event_name, count: count, sum_value: sum_value, sum_value_business_hours: sum_value_business_hours + ).each do |metric, metric_data| + key = [dimension[:type], dimension_id, metric] + aggregates[key][:count] += metric_data[:count] + aggregates[key][:sum_value] += metric_data[:sum_value].to_f + aggregates[key][:sum_value_business_hours] += metric_data[:sum_value_business_hours].to_f + end end end - def grouped_event_attributes(grouped_row) - event_name, user_id, inbox_id, count, sum_value, sum_value_business_hours = grouped_row + def aggregate_distinct_events(aggregates, scope, dimension) + return if DISTINCT_COUNT_EVENTS.empty? - { - event_name: event_name, - user_id: user_id, - inbox_id: inbox_id, - count: count, - sum_value: sum_value, - sum_value_business_hours: sum_value_business_hours - } - end + group_cols = dimension[:group_column] ? [:name, dimension[:group_column]] : [:name] - def accumulate_metric_aggregates(aggregates, dimensions, metric, metric_data) - dimensions.each do |dimension_type, dimension_id| + scope.group(*group_cols).pluck(*group_cols, DISTINCT_COUNT_SQL).each do |row| + event_name, dimension_id, count = dimension[:group_column] ? row : [row[0], @account.id, row[1]] next if dimension_id.nil? - key = [dimension_type, dimension_id, metric] - aggregates[key][:count] += metric_data[:count] - aggregates[key][:sum_value] += metric_data[:sum_value].to_f - aggregates[key][:sum_value_business_hours] += metric_data[:sum_value_business_hours].to_f + ReportingEvents::EventMetricRegistry.metrics_for_aggregate( + event_name, count: count, sum_value: 0, sum_value_business_hours: 0 + ).each do |metric, metric_data| + key = [dimension[:type], dimension_id, metric] + aggregates[key][:count] += metric_data[:count] + end + end + end + + def dimension_groups_and_selects(dimension) + agg_selects = [Arel.sql('COUNT(*)'), Arel.sql('COALESCE(SUM(value), 0)'), Arel.sql('COALESCE(SUM(value_in_business_hours), 0)')] + + if dimension[:group_column] + [[:name, dimension[:group_column]], [:name, dimension[:group_column], *agg_selects]] + else + [[:name], [:name, *agg_selects]] + end + end + + def unpack_row(row, dimension) + if dimension[:group_column] + # [name, dimension_id, count, sum_value, sum_value_business_hours] + row + else + # [name, count, sum_value, sum_value_business_hours] → inject account id + [row[0], @account.id, row[1], row[2], row[3]] end end diff --git a/spec/services/reporting_events/backfill_service_spec.rb b/spec/services/reporting_events/backfill_service_spec.rb index c74b32860..8f0572d3b 100644 --- a/spec/services/reporting_events/backfill_service_spec.rb +++ b/spec/services/reporting_events/backfill_service_spec.rb @@ -62,15 +62,72 @@ describe ReportingEvents::BackfillService do expect(reporting_event_instantiations).to eq(0) - first_response_rollup = find_rollup('agent', user.id, 'first_response') - expect(first_response_rollup.count).to eq(2) - expect(first_response_rollup.sum_value).to eq(140) - expect(first_response_rollup.sum_value_business_hours).to eq(80) + rollups = ReportingEventsRollup.where(account_id: account.id, date: date) + # 3 dimensions × first_response + 3 dimensions × resolutions_count + 3 dimensions × resolution_time + expect(rollups.count).to eq(9) - resolution_time_rollup = find_rollup('agent', second_user.id, 'resolution_time') - expect(resolution_time_rollup.count).to eq(1) - expect(resolution_time_rollup.sum_value).to eq(200) - expect(resolution_time_rollup.sum_value_business_hours).to eq(80) + # account dimension + account_first_response = find_rollup('account', account.id, 'first_response') + expect(account_first_response.count).to eq(2) + expect(account_first_response.sum_value).to eq(140) + expect(account_first_response.sum_value_business_hours).to eq(80) + + # agent dimension + agent_first_response = find_rollup('agent', user.id, 'first_response') + expect(agent_first_response.count).to eq(2) + expect(agent_first_response.sum_value).to eq(140) + expect(agent_first_response.sum_value_business_hours).to eq(80) + + agent_resolution_time = find_rollup('agent', second_user.id, 'resolution_time') + expect(agent_resolution_time.count).to eq(1) + expect(agent_resolution_time.sum_value).to eq(200) + expect(agent_resolution_time.sum_value_business_hours).to eq(80) + + # inbox dimension + inbox_first_response = find_rollup('inbox', inbox.id, 'first_response') + expect(inbox_first_response.count).to eq(2) + expect(inbox_first_response.sum_value).to eq(140) + expect(inbox_first_response.sum_value_business_hours).to eq(80) + + inbox_resolution_time = find_rollup('inbox', second_inbox.id, 'resolution_time') + expect(inbox_resolution_time.count).to eq(1) + expect(inbox_resolution_time.sum_value).to eq(200) + expect(inbox_resolution_time.sum_value_business_hours).to eq(80) + end + + it 'deduplicates distinct-count events per dimension' do + second_user = create(:user, account: account) + second_inbox = create(:inbox, account: account) + conversation_b = create(:conversation, account: account, inbox: inbox, assignee: user) + conversation_c = create(:conversation, account: account, inbox: second_inbox, assignee: second_user) + + # Two events for the same conversation — should count as 1 + create_backfill_event(name: 'conversation_bot_handoff', value: 0, value_in_business_hours: 0, user: user, + inbox: inbox, conversation: conversation, created_at: Time.utc(2026, 2, 11, 14)) + create_backfill_event(name: 'conversation_bot_handoff', value: 0, value_in_business_hours: 0, user: user, + inbox: inbox, conversation: conversation, created_at: Time.utc(2026, 2, 11, 15)) + # Different conversation, same agent/inbox + create_backfill_event(name: 'conversation_bot_handoff', value: 0, value_in_business_hours: 0, user: user, + inbox: inbox, conversation: conversation_b, created_at: Time.utc(2026, 2, 11, 16)) + # Different agent/inbox + create_backfill_event(name: 'conversation_bot_handoff', value: 0, value_in_business_hours: 0, user: second_user, + inbox: second_inbox, conversation: conversation_c, created_at: Time.utc(2026, 2, 11, 17)) + + described_class.backfill_date(account, date) + + rollups = ReportingEventsRollup.where(account_id: account.id, date: date) + expect(rollups.count).to eq(5) + + # account: 3 distinct conversations + expect(find_rollup('account', account.id, 'bot_handoffs_count').count).to eq(3) + + # agent: user has 2 distinct, second_user has 1 + expect(find_rollup('agent', user.id, 'bot_handoffs_count').count).to eq(2) + expect(find_rollup('agent', second_user.id, 'bot_handoffs_count').count).to eq(1) + + # inbox: inbox has 2 distinct, second_inbox has 1 + expect(find_rollup('inbox', inbox.id, 'bot_handoffs_count').count).to eq(2) + expect(find_rollup('inbox', second_inbox.id, 'bot_handoffs_count').count).to eq(1) end def create_backfill_event(**attributes)