feat: add backfill service for historical rollup data

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

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