refactor: aggregate backfill rollups in sql

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?

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

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

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(