diff --git a/app/services/reporting_events/backfill_service.rb b/app/services/reporting_events/backfill_service.rb index 2db0a527a..7f1dd705e 100644 --- a/app/services/reporting_events/backfill_service.rb +++ b/app/services/reporting_events/backfill_service.rb @@ -1,6 +1,15 @@ # 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 + def self.backfill_date(account, date) new(account, date).perform end @@ -11,21 +20,9 @@ class ReportingEvents::BackfillService end def perform - # 1. Delete existing rollups for this date (idempotency) delete_existing_rollups - - # 2. Convert date in account timezone to UTC boundaries start_utc, end_utc = date_boundaries_in_utc - - # 3. Process each event type with SQL aggregation - rollup_rows = [] - rollup_rows.concat(aggregate_event_type('conversation_resolved', start_utc, end_utc)) - rollup_rows.concat(aggregate_event_type('first_response', start_utc, end_utc)) - rollup_rows.concat(aggregate_event_type('reply_time', start_utc, end_utc)) - rollup_rows.concat(aggregate_event_type('conversation_bot_resolved', start_utc, end_utc)) - rollup_rows.concat(aggregate_event_type('conversation_bot_handoff', start_utc, end_utc)) - - # 4. Bulk insert all rollups at once + rollup_rows = build_rollup_rows(start_utc, end_utc) bulk_insert_rollups(rollup_rows) if rollup_rows.any? end @@ -42,16 +39,9 @@ class ReportingEvents::BackfillService [start_in_tz.utc, end_in_tz.utc] end - def aggregate_event_type(event_name, start_utc, end_utc) - events = @account.reporting_events - .where(name: event_name, created_at: start_utc...end_utc) + def build_rollup_rows(start_utc, end_utc) + aggregates = build_aggregates(start_utc, end_utc) - return [] if events.empty? - - # Build in-memory aggregates by dimension - aggregates = build_aggregates(events) - - # Convert to rollup row hashes aggregates.map do |(dimension_type, dimension_id, metric), data| { account_id: @account.id, @@ -68,30 +58,54 @@ class ReportingEvents::BackfillService end end - def build_aggregates(events) + 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 } } - events.each { |event| accumulate_event_aggregates(aggregates, event) } + grouped_events(start_utc, end_utc).each { |grouped_event| accumulate_grouped_aggregates(aggregates, grouped_event) } aggregates end - def dimensions_for_event(event) + def grouped_events(start_utc, end_utc) + @account.reporting_events + .where(name: ReportingEvents::MetricRegistry::EVENT_METRICS.keys, created_at: start_utc...end_utc) + .group(:name, :user_id, :inbox_id) + .pluck(*AGGREGATE_SELECTS) + .map { |grouped_row| grouped_event_attributes(grouped_row) } + end + + def dimensions(grouped_event) { 'account' => @account.id, - 'agent' => event.user_id, - 'inbox' => event.inbox_id + 'agent' => grouped_event[:user_id], + 'inbox' => grouped_event[:inbox_id] } end - def accumulate_event_aggregates(aggregates, event) - dimensions = dimensions_for_event(event) - - ReportingEvents::MetricRegistry.event_metrics_for(event).each do |metric, metric_data| - accumulate_metric_aggregates(aggregates, dimensions, metric, metric_data) + def accumulate_grouped_aggregates(aggregates, grouped_event) + ReportingEvents::MetricRegistry.event_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) end end + def grouped_event_attributes(grouped_row) + event_name, user_id, inbox_id, count, sum_value, sum_value_business_hours = grouped_row + + { + 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 + def accumulate_metric_aggregates(aggregates, dimensions, metric, metric_data) dimensions.each do |dimension_type, dimension_id| next if dimension_id.nil? diff --git a/app/services/reporting_events/metric_registry.rb b/app/services/reporting_events/metric_registry.rb index c0ed93b9b..1213c0c1f 100644 --- a/app/services/reporting_events/metric_registry.rb +++ b/app/services/reporting_events/metric_registry.rb @@ -5,16 +5,16 @@ # aligned. module ReportingEvents::MetricRegistry EVENT_METRICS = { - 'conversation_resolved' => lambda do |event| + 'conversation_resolved' => lambda do |values| { - resolutions_count: count_metric, - resolution_time: duration_metric(event) + resolutions_count: count_metric(values[:count]), + resolution_time: duration_metric(values) } end, - 'first_response' => ->(event) { { first_response: duration_metric(event) } }, - 'reply_time' => ->(event) { { reply_time: duration_metric(event) } }, - 'conversation_bot_resolved' => ->(_event) { { bot_resolutions_count: count_metric } }, - 'conversation_bot_handoff' => ->(_event) { { bot_handoffs_count: count_metric } } + 'first_response' => ->(values) { { first_response: duration_metric(values) } }, + 'reply_time' => ->(values) { { reply_time: duration_metric(values) } }, + 'conversation_bot_resolved' => ->(values) { { bot_resolutions_count: count_metric(values[:count]) } }, + 'conversation_bot_handoff' => ->(values) { { bot_handoffs_count: count_metric(values[:count]) } } }.freeze REPORT_METRICS = { @@ -64,8 +64,24 @@ module ReportingEvents::MetricRegistry def event_metrics_for(event) return {} if event.blank? + return {} unless EVENT_METRICS.key?(event.name.to_s) - EVENT_METRICS[event.name.to_s]&.call(event) || {} + event_metrics_for_aggregate( + event.name, + count: 1, + sum_value: event.try(:value), + sum_value_business_hours: event.try(:value_in_business_hours) + ) + end + + def event_metrics_for_aggregate(event_name, count:, sum_value:, sum_value_business_hours:) + values = { + count: count.to_i, + sum_value: sum_value.to_f, + sum_value_business_hours: sum_value_business_hours.to_f + } + + EVENT_METRICS[event_name.to_s]&.call(values) || {} end def report_metric(metric) @@ -74,36 +90,26 @@ module ReportingEvents::MetricRegistry REPORT_METRICS[metric.to_sym] end - def supported_metric?(metric) - report_metric(metric).present? - end + def supported_metric?(metric) = report_metric(metric).present? - def aggregate_for(metric) - report_metric(metric)&.dig(:aggregate) - end + def aggregate_for(metric) = report_metric(metric)&.dig(:aggregate) - def rollup_supported_metric?(metric) - rollup_metric_for(metric).present? - end + def rollup_supported_metric?(metric) = rollup_metric_for(metric).present? - def rollup_metric_for(metric) - report_metric(metric)&.dig(:rollup_metric) - end + def rollup_metric_for(metric) = report_metric(metric)&.dig(:rollup_metric) - def raw_event_name_for(metric) - report_metric(metric)&.dig(:raw_event_name) - end + def raw_event_name_for(metric) = report_metric(metric)&.dig(:raw_event_name) - def count_metric - { count: 1, sum_value: 0, sum_value_business_hours: 0 } + def count_metric(count) + { count: count, sum_value: 0, sum_value_business_hours: 0 } end private_class_method :count_metric - def duration_metric(event) + def duration_metric(values) { - count: 1, - sum_value: event.value.to_f, - sum_value_business_hours: event.value_in_business_hours.to_f + count: values[:count], + sum_value: values[:sum_value], + sum_value_business_hours: values[:sum_value_business_hours] } end private_class_method :duration_metric diff --git a/spec/services/reporting_events/backfill_service_spec.rb b/spec/services/reporting_events/backfill_service_spec.rb index 228f59c08..c74b32860 100644 --- a/spec/services/reporting_events/backfill_service_spec.rb +++ b/spec/services/reporting_events/backfill_service_spec.rb @@ -41,5 +41,67 @@ describe ReportingEvents::BackfillService do expect(rollup.sum_value).to eq(0) expect(rollup.sum_value_business_hours).to eq(0) end + + it 'aggregates grouped rows without instantiating reporting events' do + second_user = create(:user, account: account) + second_inbox = create(:inbox, account: account) + second_conversation = create(:conversation, account: account, inbox: second_inbox, assignee: second_user) + + create_backfill_event(name: 'first_response', value: 100, value_in_business_hours: 60, user: user, + inbox: inbox, conversation: conversation, created_at: Time.utc(2026, 2, 11, 14)) + create_backfill_event(name: 'first_response', value: 40, value_in_business_hours: 20, user: user, + inbox: inbox, conversation: conversation, created_at: Time.utc(2026, 2, 11, 15)) + create_backfill_event(name: 'conversation_resolved', value: 200, value_in_business_hours: 80, user: second_user, + inbox: second_inbox, conversation: second_conversation, created_at: Time.utc(2026, 2, 11, 16)) + create_backfill_event(name: 'reply_time', value: 500, value_in_business_hours: 300, user: user, + inbox: inbox, conversation: conversation, created_at: Time.utc(2026, 2, 12, 5)) + + reporting_event_instantiations = count_reporting_event_instantiations do + described_class.backfill_date(account, date) + end + + 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) + + 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) + end + + def create_backfill_event(**attributes) + create( + :reporting_event, + account: account, + **attributes + ) + end + + def find_rollup(dimension_type, dimension_id, metric) + ReportingEventsRollup.find_by!( + account_id: account.id, + date: date, + dimension_type: dimension_type, + dimension_id: dimension_id, + metric: metric + ) + end + + def count_reporting_event_instantiations(&) + instantiation_count = 0 + subscriber = lambda do |_name, _start, _finish, _id, payload| + next unless payload[:class_name] == 'ReportingEvent' + + instantiation_count += payload[:record_count] + end + + ActiveSupport::Notifications.subscribed(subscriber, 'instantiation.active_record', &) + + instantiation_count + end end end diff --git a/spec/services/reporting_events/metric_registry_spec.rb b/spec/services/reporting_events/metric_registry_spec.rb index 3a4b8d0c1..22357330c 100644 --- a/spec/services/reporting_events/metric_registry_spec.rb +++ b/spec/services/reporting_events/metric_registry_spec.rb @@ -74,6 +74,41 @@ RSpec.describe ReportingEvents::MetricRegistry do end end + describe '.event_metrics_for_aggregate' do + it 'returns aggregated rollup metrics for conversation_resolved groups' do + expect( + described_class.event_metrics_for_aggregate( + 'conversation_resolved', + count: 3, + sum_value: 420, + sum_value_business_hours: 210 + ) + ).to eq( + resolutions_count: { + count: 3, + sum_value: 0, + sum_value_business_hours: 0 + }, + resolution_time: { + count: 3, + sum_value: 420.0, + sum_value_business_hours: 210.0 + } + ) + end + + it 'returns an empty hash for unsupported grouped events' do + expect( + described_class.event_metrics_for_aggregate( + 'conversation_created', + count: 2, + sum_value: 100, + sum_value_business_hours: 50 + ) + ).to eq({}) + end + end + describe '.report_metric' do it 'returns the definition for raw-only count metrics' do expect(described_class.report_metric(:conversations_count)).to eq(