diff --git a/app/services/reporting_events/backfill_service.rb b/app/services/reporting_events/backfill_service.rb new file mode 100644 index 000000000..7f1dd705e --- /dev/null +++ b/app/services/reporting_events/backfill_service.rb @@ -0,0 +1,125 @@ +# 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 + + def initialize(account, date) + @account = account + @date = date + end + + def perform + delete_existing_rollups + start_utc, end_utc = date_boundaries_in_utc + rollup_rows = build_rollup_rows(start_utc, end_utc) + bulk_insert_rollups(rollup_rows) if rollup_rows.any? + end + + private + + def delete_existing_rollups + ReportingEventsRollup.where(account_id: @account.id, date: @date).delete_all + end + + def date_boundaries_in_utc + tz = ActiveSupport::TimeZone[@account.reporting_timezone] + start_in_tz = tz.parse(@date.to_s) + end_in_tz = start_in_tz + 1.day + [start_in_tz.utc, end_in_tz.utc] + end + + def build_rollup_rows(start_utc, end_utc) + aggregates = build_aggregates(start_utc, end_utc) + + aggregates.map do |(dimension_type, dimension_id, metric), data| + { + account_id: @account.id, + date: @date, + dimension_type: dimension_type, + dimension_id: dimension_id, + metric: metric, + count: data[:count], + sum_value: data[:sum_value], + sum_value_business_hours: data[:sum_value_business_hours], + created_at: Time.current, + updated_at: Time.current + } + end + end + + 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 } } + + grouped_events(start_utc, end_utc).each { |grouped_event| accumulate_grouped_aggregates(aggregates, grouped_event) } + + aggregates + end + + 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' => grouped_event[:user_id], + 'inbox' => grouped_event[:inbox_id] + } + end + + 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? + + 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 + + def bulk_insert_rollups(rollup_rows) + # rubocop:disable Rails/SkipsModelValidations + ReportingEventsRollup.insert_all(rollup_rows) + # rubocop:enable Rails/SkipsModelValidations + end +end diff --git a/spec/services/reporting_events/backfill_service_spec.rb b/spec/services/reporting_events/backfill_service_spec.rb new file mode 100644 index 000000000..c74b32860 --- /dev/null +++ b/spec/services/reporting_events/backfill_service_spec.rb @@ -0,0 +1,107 @@ +require 'rails_helper' + +describe ReportingEvents::BackfillService do + describe '.backfill_date' do + let(:account) { create(:account, reporting_timezone: 'America/New_York') } + let(:date) { Date.new(2026, 2, 11) } + let(:user) { create(:user, account: account) } + let(:inbox) { create(:inbox, account: account) } + let(:conversation) { create(:conversation, account: account, inbox: inbox, assignee: user) } + + it 'treats nil metric values as zero during backfill' do + reporting_event = create( + :reporting_event, + account: account, + name: 'first_response', + value: 100, + value_in_business_hours: 50, + user: user, + inbox: inbox, + conversation: conversation, + created_at: Time.utc(2026, 2, 11, 15) + ) + # Simulate a legacy row that already exists in the database with nil metrics. + # rubocop:disable Rails/SkipsModelValidations + reporting_event.update_columns(value: nil, value_in_business_hours: nil) + # rubocop:enable Rails/SkipsModelValidations + + expect do + described_class.backfill_date(account, date) + end.not_to raise_error + + rollup = ReportingEventsRollup.find_by!( + account_id: account.id, + date: date, + dimension_type: 'account', + dimension_id: account.id, + metric: 'first_response' + ) + + expect(rollup.count).to eq(1) + 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