refactor: add report data sources

app/services/reports/data_source.rb

@@ -0,0 +1,96 @@
+class Reports::DataSource
+  SUPPORTED_ROLLUP_DIMENSIONS = %w[account agent inbox].freeze
+
+  attr_reader :account, :metric, :dimension_type, :dimension_id,
+              :scope, :range, :group_by, :timezone,
+              :timezone_offset, :business_hours
+
+  class << self
+    def for(**context)
+      adapter_class_for(**context).new(**context)
+    end
+
+    def rollup_eligible?(**context)
+      account = context[:account]
+
+      rollup_enabled_for_account?(account) &&
+        !hourly_grouping?(context[:group_by]) &&
+        supported_dimension?(context[:dimension_type]) &&
+        timezone_matches_account?(account, context[:timezone_offset]) &&
+        supported_metric?(context[:metric])
+    end
+
+    def timezone_matches_account?(account, timezone_offset)
+      return false if timezone_offset.blank?
+
+      offset_in_seconds = timezone_offset.to_f * 3600
+      account_zone = ActiveSupport::TimeZone[account.reporting_timezone]
+      account_zone&.now&.utc_offset == offset_in_seconds
+    end
+
+    private
+
+    def adapter_class_for(**context)
+      rollup_eligible?(**context) ? Reports::RollupDataSource : Reports::RawDataSource
+    end
+
+    def rollup_enabled_for_account?(account)
+      account.reporting_timezone.present? && account.feature_enabled?('reporting_events_rollup')
+    end
+
+    def hourly_grouping?(group_by)
+      group_by.to_s == 'hour'
+    end
+
+    def supported_dimension?(dimension_type)
+      SUPPORTED_ROLLUP_DIMENSIONS.include?((dimension_type.presence || 'account').to_s)
+    end
+
+    def supported_metric?(metric)
+      metric.blank? || ReportingEvents::MetricRegistry.rollup_supported_metric?(metric)
+    end
+  end
+
+  def initialize(**context)
+    @account = context[:account]
+    @metric = context[:metric]
+    @dimension_type = (context[:dimension_type].presence || 'account').to_s
+    @dimension_id = context[:dimension_id]
+    @scope = context[:scope]
+    @range = context[:range]
+    @group_by = context[:group_by].to_s.presence || 'day'
+    @timezone = context[:timezone]
+    @timezone_offset = context[:timezone_offset]
+    @business_hours = context[:business_hours]
+  end
+
+  private
+
+  def report_metric
+    @report_metric ||= ReportingEvents::MetricRegistry.report_metric(metric)
+  end
+
+  def average_metric?
+    report_metric&.dig(:aggregate) == :average
+  end
+
+  def count_metric?
+    !average_metric?
+  end
+
+  def rollup_metric
+    report_metric&.dig(:rollup_metric)
+  end
+
+  def raw_event_name
+    report_metric&.dig(:raw_event_name)
+  end
+
+  def raw_count_strategy
+    report_metric&.dig(:raw_count_strategy)
+  end
+
+  def use_business_hours?
+    ActiveModel::Type::Boolean.new.cast(business_hours)
+  end
+end

app/services/reports/raw_data_source.rb

@@ -0,0 +1,143 @@
+class Reports::RawDataSource < Reports::DataSource
+  def timeseries
+    average_metric? ? average_timeseries : count_timeseries
+  end
+
+  def aggregate
+    average_metric? ? average_scope.average(average_value_key) : count_scope.count
+  end
+
+  def summary
+    results = summary_scope
+              .select(*summary_select_fields)
+              .group(summary_group_by_key)
+              .index_by { |record| record.public_send(summary_index_key) }
+
+    results.transform_values do |record|
+      {
+        resolved_conversations_count: record.resolved_count.to_i,
+        avg_resolution_time: record.avg_resolution_time,
+        avg_first_response_time: record.avg_first_response_time,
+        avg_reply_time: record.avg_reply_time
+      }
+    end
+  end
+
+  private
+
+  def count_timeseries
+    grouped_count.map do |event_date, event_count|
+      { value: event_count, timestamp: event_date.in_time_zone(timezone).to_i }
+    end
+  end
+
+  def average_timeseries
+    grouped_average_time = grouped_average_scope.average(average_value_key)
+    grouped_event_count = grouped_average_scope.count
+
+    grouped_average_time.each_with_object([]) do |(event_date, average_time), results|
+      results << {
+        value: average_time,
+        timestamp: event_date.in_time_zone(timezone).to_i,
+        count: grouped_event_count[event_date]
+      }
+    end
+  end
+
+  def grouped_average_scope
+    average_scope.group_by_period(
+      group_by,
+      :created_at,
+      default_value: 0,
+      range: range,
+      permit: %w[day week month year hour],
+      time_zone: timezone
+    )
+  end
+
+  def grouped_count
+    count_scope.group_by_period(
+      group_by,
+      :created_at,
+      default_value: 0,
+      range: range,
+      permit: %w[day week month year hour],
+      time_zone: timezone
+    ).count
+  end
+
+  def average_scope
+    scope.reporting_events.where(name: raw_event_name, created_at: range, account_id: account.id)
+  end
+
+  def count_scope
+    case metric.to_s
+    when 'conversations_count'
+      scope.conversations.where(account_id: account.id, created_at: range)
+    when 'incoming_messages_count'
+      scope.messages.where(account_id: account.id, created_at: range).incoming.unscope(:order)
+    when 'outgoing_messages_count'
+      scope.messages.where(account_id: account.id, created_at: range).outgoing.unscope(:order)
+    else
+      reporting_event_count_scope
+    end
+  end
+
+  def reporting_event_count_scope
+    events = scope.reporting_events.where(
+      name: raw_event_name,
+      account_id: account.id,
+      created_at: range
+    )
+
+    return events unless raw_count_strategy == :distinct_conversation
+
+    events.joins(:conversation).select(:conversation_id).distinct
+  end
+
+  def summary_scope
+    scope = account.reporting_events.where(created_at: range)
+    return scope.joins(:conversation) if dimension_type == 'team'
+
+    scope
+  end
+
+  def summary_select_fields
+    [
+      "#{summary_group_by_key} as #{summary_index_key}",
+      count_select(:resolutions_count, :resolved_count),
+      average_select(:avg_resolution_time, :avg_resolution_time),
+      average_select(:avg_first_response_time, :avg_first_response_time),
+      average_select(:reply_time, :avg_reply_time)
+    ]
+  end
+
+  def count_select(metric_name, alias_name)
+    "COUNT(CASE WHEN name = '#{raw_metric_event_name(metric_name)}' THEN 1 END) as #{alias_name}"
+  end
+
+  def average_select(metric_name, alias_name)
+    "AVG(CASE WHEN name = '#{raw_metric_event_name(metric_name)}' THEN #{average_value_key} END) as #{alias_name}"
+  end
+
+  def raw_metric_event_name(metric_name)
+    ReportingEvents::MetricRegistry.raw_event_name_for(metric_name)
+  end
+
+  def summary_group_by_key
+    {
+      'account' => :account_id,
+      'agent' => :user_id,
+      'inbox' => :inbox_id,
+      'team' => 'conversations.team_id'
+    }[dimension_type]
+  end
+
+  def summary_index_key
+    summary_group_by_key.to_s.split('.').last
+  end
+
+  def average_value_key
+    use_business_hours? ? :value_in_business_hours : :value
+  end
+end

app/services/reports/rollup_data_source.rb

@@ -0,0 +1,170 @@
+class Reports::RollupDataSource < Reports::DataSource
+  def timeseries
+    count_metric? ? count_timeseries : average_timeseries
+  end
+
+  def aggregate
+    count_metric? ? count_aggregate : average_aggregate
+  end
+
+  def summary
+    summary_rows.each_with_object({}) do |row, results|
+      results[row.dimension_id] = summary_attributes_for(row)
+    end
+  end
+
+  private
+
+  def count_timeseries
+    grouped_data = all_periods_in_range.index_with { 0 }
+
+    rollup_scope.each do |row|
+      date_key = normalized_period_key(row.date)
+      grouped_data[date_key] ||= 0
+      grouped_data[date_key] += row.count
+    end
+
+    results = grouped_data.map do |date_key, count|
+      { value: count, timestamp: date_key.in_time_zone(timezone).to_i }
+    end
+
+    results.sort_by { |result| result[:timestamp] }
+  end
+
+  def average_timeseries
+    grouped_data = all_periods_in_range.index_with { { count: 0, sum_value: 0.0 } }
+
+    rollup_scope.each { |row| accumulate_average_row(grouped_data, row) }
+
+    results = grouped_data.map do |date_key, data|
+      {
+        value: data[:count].zero? ? 0 : data[:sum_value] / data[:count],
+        timestamp: date_key.in_time_zone(timezone).to_i,
+        count: data[:count]
+      }
+    end
+
+    results.sort_by { |result| result[:timestamp] }
+  end
+
+  def count_aggregate
+    rollup_scope.sum(:count).to_i
+  end
+
+  def average_aggregate
+    result = rollup_scope.pick(Arel.sql("SUM(count), SUM(#{rollup_value_column})"))
+    return nil if result.blank? || result[0].to_i.zero?
+
+    result[1].to_f / result[0].to_i
+  end
+
+  def rollup_scope
+    ReportingEventsRollup.where(
+      account_id: account.id,
+      metric: rollup_metric,
+      dimension_type: dimension_type,
+      dimension_id: dimension_id_for_rollup,
+      date: rollup_date_range
+    )
+  end
+
+  def summary_rows
+    ReportingEventsRollup.where(
+      account_id: account.id,
+      dimension_type: dimension_type,
+      date: rollup_date_range
+    ).group(:dimension_id).select(*summary_select_fields)
+  end
+
+  def summary_select_fields
+    [
+      'dimension_id',
+      sum_count_select(:resolutions_count, :resolved_count),
+      sum_count_select(:avg_resolution_time, :resolution_count),
+      sum_value_select(:avg_resolution_time, :resolution_sum_value),
+      sum_count_select(:avg_first_response_time, :first_response_count),
+      sum_value_select(:avg_first_response_time, :first_response_sum_value),
+      sum_count_select(:reply_time, :reply_count),
+      sum_value_select(:reply_time, :reply_sum_value)
+    ]
+  end
+
+  def sum_count_select(metric_name, alias_name)
+    "SUM(CASE WHEN metric = '#{rollup_metric_name(metric_name)}' THEN count ELSE 0 END) as #{alias_name}"
+  end
+
+  def sum_value_select(metric_name, alias_name)
+    "SUM(CASE WHEN metric = '#{rollup_metric_name(metric_name)}' THEN #{rollup_value_column} ELSE 0 END) as #{alias_name}"
+  end
+
+  def rollup_metric_name(metric_name)
+    ReportingEvents::MetricRegistry.rollup_metric_for(metric_name)
+  end
+
+  def summary_attributes_for(row)
+    {
+      resolved_conversations_count: row.resolved_count.to_i,
+      avg_resolution_time: average_from(row.resolution_sum_value, row.resolution_count),
+      avg_first_response_time: average_from(row.first_response_sum_value, row.first_response_count),
+      avg_reply_time: average_from(row.reply_sum_value, row.reply_count)
+    }
+  end
+
+  def dimension_id_for_rollup
+    dimension_type == 'account' ? account.id : scope.id
+  end
+
+  def rollup_value_column
+    use_business_hours? ? :sum_value_business_hours : :sum_value
+  end
+
+  def rollup_date_range
+    tz = ActiveSupport::TimeZone[account.reporting_timezone]
+    start_date = range.first.in_time_zone(tz).to_date
+    end_date = (range.last - 1.second).in_time_zone(tz).to_date
+    start_date..end_date
+  end
+
+  def all_periods_in_range
+    current = normalized_period_key(rollup_date_range.first)
+    periods = []
+
+    while current <= rollup_date_range.last
+      periods << current
+      current = advance_period(current)
+    end
+
+    periods
+  end
+
+  def accumulate_average_row(grouped_data, row)
+    date_key = normalized_period_key(row.date)
+    grouped_data[date_key] ||= { count: 0, sum_value: 0.0 }
+    grouped_data[date_key][:count] += row.count
+    grouped_data[date_key][:sum_value] += row.public_send(rollup_value_column)
+  end
+
+  def normalized_period_key(date)
+    case group_by
+    when 'week' then date.beginning_of_week(:sunday)
+    when 'month' then date.beginning_of_month
+    when 'year' then date.beginning_of_year
+    else date
+    end
+  end
+
+  def advance_period(date)
+    case group_by
+    when 'week' then date + 1.week
+    when 'month' then date + 1.month
+    when 'year' then date + 1.year
+    else date + 1.day
+    end
+  end
+
+  def average_from(sum_value, count)
+    return nil if count.to_i.zero?
+
+    sum_value.to_f / count.to_i
+  end
+end

spec/services/reports/data_source_adapter_contract_spec.rb

@@ -0,0 +1,268 @@
+require 'rails_helper'
+
+RSpec.describe Reports::DataSource do
+  let(:account) { create(:account, reporting_timezone: 'Etc/UTC') }
+  let(:user1) { create(:user) }
+  let(:user2) { create(:user) }
+  let(:inbox1) { create(:inbox, account: account) }
+  let(:inbox2) { create(:inbox, account: account) }
+  let(:timezone) { 'UTC' }
+  let(:timezone_offset) { '0' }
+  let(:current_time) { Time.zone.parse('2026-01-15 10:00:00 UTC') }
+  let(:full_range) { Time.zone.parse('2026-01-05 00:00:00 UTC')...Time.zone.parse('2026-01-16 00:00:00 UTC') }
+  let(:day_range) { Time.zone.parse('2026-01-14 00:00:00 UTC')...Time.zone.parse('2026-01-16 00:00:00 UTC') }
+
+  before do
+    travel_to current_time
+    create(:account_user, account: account, user: user1)
+    create(:account_user, account: account, user: user2)
+
+    conversation_one = create(
+      :conversation,
+      account: account,
+      inbox: inbox1,
+      assignee: user1,
+      created_at: Time.zone.parse('2026-01-06 09:00:00 UTC')
+    )
+    conversation_two = create(
+      :conversation,
+      account: account,
+      inbox: inbox1,
+      assignee: user1,
+      created_at: Time.zone.parse('2026-01-14 09:00:00 UTC')
+    )
+    conversation_three = create(
+      :conversation,
+      account: account,
+      inbox: inbox2,
+      assignee: user2,
+      created_at: Time.zone.parse('2026-01-15 09:00:00 UTC')
+    )
+
+    create(:reporting_event,
+           name: 'first_response',
+           account: account,
+           user: user1,
+           inbox: inbox1,
+           conversation: conversation_one,
+           value: 20,
+           value_in_business_hours: 10,
+           created_at: Time.zone.parse('2026-01-06 10:00:00 UTC'))
+    create(:reporting_event,
+           name: 'reply_time',
+           account: account,
+           user: user1,
+           inbox: inbox1,
+           conversation: conversation_one,
+           value: 10,
+           value_in_business_hours: 5,
+           created_at: Time.zone.parse('2026-01-06 11:00:00 UTC'))
+    create(:reporting_event,
+           name: 'conversation_resolved',
+           account: account,
+           user: user1,
+           inbox: inbox1,
+           conversation: conversation_one,
+           value: 100,
+           value_in_business_hours: 80,
+           created_at: Time.zone.parse('2026-01-06 12:00:00 UTC'))
+
+    create(:reporting_event,
+           name: 'first_response',
+           account: account,
+           user: user1,
+           inbox: inbox1,
+           conversation: conversation_two,
+           value: 40,
+           value_in_business_hours: 30,
+           created_at: Time.zone.parse('2026-01-14 10:00:00 UTC'))
+    create(:reporting_event,
+           name: 'reply_time',
+           account: account,
+           user: user1,
+           inbox: inbox1,
+           conversation: conversation_two,
+           value: 30,
+           value_in_business_hours: 20,
+           created_at: Time.zone.parse('2026-01-14 11:00:00 UTC'))
+    create(:reporting_event,
+           name: 'conversation_resolved',
+           account: account,
+           user: user1,
+           inbox: inbox1,
+           conversation: conversation_two,
+           value: 200,
+           value_in_business_hours: 150,
+           created_at: Time.zone.parse('2026-01-14 12:00:00 UTC'))
+
+    create(:reporting_event,
+           name: 'first_response',
+           account: account,
+           user: user2,
+           inbox: inbox2,
+           conversation: conversation_three,
+           value: 60,
+           value_in_business_hours: 50,
+           created_at: Time.zone.parse('2026-01-15 10:00:00 UTC'))
+    create(:reporting_event,
+           name: 'reply_time',
+           account: account,
+           user: user2,
+           inbox: inbox2,
+           conversation: conversation_three,
+           value: 50,
+           value_in_business_hours: 40,
+           created_at: Time.zone.parse('2026-01-15 11:00:00 UTC'))
+    create(:reporting_event,
+           name: 'conversation_resolved',
+           account: account,
+           user: user2,
+           inbox: inbox2,
+           conversation: conversation_three,
+           value: 300,
+           value_in_business_hours: 250,
+           created_at: Time.zone.parse('2026-01-15 12:00:00 UTC'))
+
+    [Date.new(2026, 1, 6), Date.new(2026, 1, 14), Date.new(2026, 1, 15)].each do |date|
+      ReportingEvents::BackfillService.backfill_date(account, date)
+    end
+  end
+
+  shared_examples 'report adapter contract' do |adapter_class|
+    context 'with average metrics' do
+      subject(:source) do
+        adapter_class.new(
+          account: account,
+          metric: 'avg_first_response_time',
+          dimension_type: 'account',
+          dimension_id: nil,
+          scope: account,
+          range: day_range,
+          group_by: 'day',
+          timezone: timezone,
+          timezone_offset: timezone_offset,
+          business_hours: business_hours
+        )
+      end
+
+      let(:business_hours) { false }
+
+      it 'returns the expected aggregate' do
+        expect(source.aggregate).to eq(50.0)
+      end
+
+      it 'returns the expected day timeseries' do
+        expect(source.timeseries).to eq(
+          [
+            { count: 1, timestamp: Date.new(2026, 1, 14).in_time_zone(timezone).to_i, value: 40.0 },
+            { count: 1, timestamp: Date.new(2026, 1, 15).in_time_zone(timezone).to_i, value: 60.0 }
+          ]
+        )
+      end
+
+      context 'when business hours are requested' do
+        let(:business_hours) { true }
+
+        it 'returns the business-hours aggregate and timeseries' do
+          expect(source.aggregate).to eq(40.0)
+          expect(source.timeseries).to eq(
+            [
+              { count: 1, timestamp: Date.new(2026, 1, 14).in_time_zone(timezone).to_i, value: 30.0 },
+              { count: 1, timestamp: Date.new(2026, 1, 15).in_time_zone(timezone).to_i, value: 50.0 }
+            ]
+          )
+        end
+      end
+    end
+
+    context 'with count metrics' do
+      subject(:source) do
+        adapter_class.new(
+          account: account,
+          metric: 'resolutions_count',
+          dimension_type: 'agent',
+          dimension_id: user1.id,
+          scope: user1,
+          range: full_range,
+          group_by: 'week',
+          timezone: timezone,
+          timezone_offset: timezone_offset,
+          business_hours: false
+        )
+      end
+
+      it 'returns the expected aggregate' do
+        expect(source.aggregate).to eq(2)
+      end
+
+      it 'returns the expected week timeseries' do
+        expect(source.timeseries).to eq(
+          [
+            { value: 1, timestamp: Date.new(2026, 1, 4).in_time_zone(timezone).to_i },
+            { value: 1, timestamp: Date.new(2026, 1, 11).in_time_zone(timezone).to_i }
+          ]
+        )
+      end
+    end
+
+    context 'with summary metrics' do
+      subject(:source) do
+        adapter_class.new(
+          account: account,
+          metric: nil,
+          dimension_type: 'inbox',
+          dimension_id: nil,
+          scope: nil,
+          range: full_range,
+          group_by: 'day',
+          timezone: timezone,
+          timezone_offset: timezone_offset,
+          business_hours: business_hours
+        )
+      end
+
+      let(:business_hours) { false }
+
+      it 'returns the expected summary shape and values' do
+        expect(source.summary).to eq(
+          inbox1.id => {
+            resolved_conversations_count: 2,
+            avg_resolution_time: 150.0,
+            avg_first_response_time: 30.0,
+            avg_reply_time: 20.0
+          },
+          inbox2.id => {
+            resolved_conversations_count: 1,
+            avg_resolution_time: 300.0,
+            avg_first_response_time: 60.0,
+            avg_reply_time: 50.0
+          }
+        )
+      end
+
+      context 'when business hours are requested' do
+        let(:business_hours) { true }
+
+        it 'returns the business-hours summary values' do
+          expect(source.summary).to eq(
+            inbox1.id => {
+              resolved_conversations_count: 2,
+              avg_resolution_time: 115.0,
+              avg_first_response_time: 20.0,
+              avg_reply_time: 12.5
+            },
+            inbox2.id => {
+              resolved_conversations_count: 1,
+              avg_resolution_time: 250.0,
+              avg_first_response_time: 50.0,
+              avg_reply_time: 40.0
+            }
+          )
+        end
+      end
+    end
+  end
+
+  it_behaves_like 'report adapter contract', Reports::RawDataSource
+  it_behaves_like 'report adapter contract', Reports::RollupDataSource
+end

spec/services/reports/data_source_spec.rb

@@ -0,0 +1,62 @@
+require 'rails_helper'
+
+RSpec.describe Reports::DataSource do
+  describe '.for' do
+    let(:account) { create(:account, reporting_timezone: 'America/New_York') }
+    let(:current_offset) { ActiveSupport::TimeZone['America/New_York'].now.utc_offset / 3600.0 }
+    let(:params) do
+      {
+        account: account,
+        metric: 'avg_resolution_time',
+        dimension_type: 'account',
+        dimension_id: nil,
+        scope: account,
+        range: 1.day.ago.beginning_of_day...Time.current.end_of_day,
+        group_by: 'day',
+        timezone: 'America/New_York',
+        timezone_offset: current_offset,
+        business_hours: false
+      }
+    end
+
+    before do
+      allow(account).to receive(:feature_enabled?).with('reporting_events_rollup').and_return(true)
+    end
+
+    it 'returns the rollup adapter when the request is eligible' do
+      expect(described_class.for(**params)).to be_a(Reports::RollupDataSource)
+    end
+
+    it 'falls back to the raw adapter when the feature flag is disabled' do
+      allow(account).to receive(:feature_enabled?).with('reporting_events_rollup').and_return(false)
+
+      expect(described_class.for(**params)).to be_a(Reports::RawDataSource)
+    end
+
+    it 'falls back to the raw adapter when the reporting timezone is missing' do
+      account.update!(reporting_timezone: nil)
+
+      expect(described_class.for(**params)).to be_a(Reports::RawDataSource)
+    end
+
+    it 'falls back to the raw adapter when the timezone offset does not match the account' do
+      expect(described_class.for(**params, timezone_offset: 0)).to be_a(Reports::RawDataSource)
+    end
+
+    it 'falls back to the raw adapter for hourly groupings' do
+      expect(described_class.for(**params, group_by: 'hour')).to be_a(Reports::RawDataSource)
+    end
+
+    it 'falls back to the raw adapter for unsupported dimensions' do
+      expect(described_class.for(**params, dimension_type: 'team')).to be_a(Reports::RawDataSource)
+    end
+
+    it 'returns the rollup adapter for summary queries without a metric' do
+      expect(described_class.for(**params, metric: nil, dimension_type: 'agent')).to be_a(Reports::RollupDataSource)
+    end
+
+    it 'falls back to the raw adapter for raw-only metrics' do
+      expect(described_class.for(**params, metric: 'conversations_count')).to be_a(Reports::RawDataSource)
+    end
+  end
+end