diff --git a/app/services/reports/rollup_data_source.rb b/app/services/reports/rollup_data_source.rb
new file mode 100644
index 000000000..765afba8a
--- /dev/null
+++ b/app/services/reports/rollup_data_source.rb
@@ -0,0 +1,199 @@
+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
+    metric_results = summary_rows.index_by(&:dimension_id)
+
+    merge_summary_results(metric_results, summary_conversation_counts)
+  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_conversation_counts
+    account.conversations
+           .where(created_at: range)
+           .group(summary_conversation_group_by_key)
+           .count
+  end
+
+  def merge_summary_results(metric_results, conversation_counts)
+    (metric_results.keys | conversation_counts.keys).index_with do |dimension_id|
+      summary_attributes_for(metric_results[dimension_id], conversation_counts[dimension_id])
+    end
+  end
+
+  def summary_select_fields
+    ['dimension_id'] + summary_metrics.flat_map { |definition| summary_select_fields_for_metric(definition) }
+  end
+
+  def summary_select_fields_for_metric(definition)
+    return [sum_count_select(definition[:rollup_metric], definition[:summary_key])] if definition[:aggregate] == :count
+
+    [
+      sum_count_select(definition[:rollup_metric], summary_count_alias(definition)),
+      sum_value_select(definition[:rollup_metric], summary_sum_alias(definition))
+    ]
+  end
+
+  def sum_count_select(rollup_metric_name, alias_name)
+    "SUM(CASE WHEN metric = '#{rollup_metric_name}' THEN count ELSE 0 END) as #{alias_name}"
+  end
+
+  def sum_value_select(rollup_metric_name, alias_name)
+    "SUM(CASE WHEN metric = '#{rollup_metric_name}' THEN #{rollup_value_column} ELSE 0 END) as #{alias_name}"
+  end
+
+  def summary_attributes_for(row, conversations_count = 0)
+    summary_metrics.each_with_object({ conversations_count: conversations_count.to_i }) do |definition, attributes|
+      attributes[definition[:summary_key]] = summary_value_for(row, definition)
+    end
+  end
+
+  def summary_value_for(row, definition)
+    return row&.public_send(definition[:summary_key]).to_i if definition[:aggregate] == :count
+
+    average_from(row&.public_send(summary_sum_alias(definition)), row&.public_send(summary_count_alias(definition)))
+  end
+
+  def summary_count_alias(definition)
+    "#{definition[:summary_key]}_count"
+  end
+
+  def summary_sum_alias(definition)
+    "#{definition[:summary_key]}_sum_value"
+  end
+
+  def dimension_id_for_rollup
+    dimension_type == 'account' ? account.id : scope.id
+  end
+
+  def summary_conversation_group_by_key
+    {
+      'account' => :account_id,
+      'agent' => :assignee_id,
+      'inbox' => :inbox_id,
+      'team' => :team_id
+    }[dimension_type]
+  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
diff --git a/spec/services/reports/data_source_spec.rb b/spec/services/reports/data_source_spec.rb
new file mode 100644
index 000000000..851bcd7a2
--- /dev/null
+++ b/spec/services/reports/data_source_spec.rb
@@ -0,0 +1,398 @@
+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 'matches rollups using the requested timezone identifier' do
+      account.update!(reporting_timezone: 'Chennai')
+
+      expect(
+        described_class.for(**params, timezone: 'Asia/Kolkata', timezone_offset: 0)
+      ).to be_a(Reports::RollupDataSource)
+    end
+
+    it 'falls back to the raw adapter when the requested timezone differs from the account' do
+      account.update!(reporting_timezone: 'Chennai')
+
+      expect(
+        described_class.for(**params, timezone: 'Asia/Colombo', timezone_offset: 5.5)
+      ).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: nil, 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
+
+  describe 'summary select fields' do
+    let(:account) { create(:account, reporting_timezone: 'Etc/UTC') }
+    let(:context) do
+      {
+        account: account,
+        metric: nil,
+        dimension_type: 'inbox',
+        dimension_id: nil,
+        scope: nil,
+        range: 1.day.ago.beginning_of_day...Time.current.end_of_day,
+        group_by: 'day',
+        timezone: 'UTC',
+        timezone_offset: '0',
+        business_hours: false
+      }
+    end
+
+    it 'derives raw summary selects from registry summary metrics' do
+      source = Reports::RawDataSource.new(**context)
+
+      expect(source.send(:summary_select_fields)).to eq(
+        [
+          'inbox_id as inbox_id',
+          "COUNT(CASE WHEN name = 'conversation_resolved' THEN 1 END) as resolved_conversations_count",
+          "AVG(CASE WHEN name = 'conversation_resolved' THEN value END) as avg_resolution_time",
+          "AVG(CASE WHEN name = 'first_response' THEN value END) as avg_first_response_time",
+          "AVG(CASE WHEN name = 'reply_time' THEN value END) as avg_reply_time"
+        ]
+      )
+    end
+
+    it 'derives rollup summary selects from registry summary metrics' do
+      source = Reports::RollupDataSource.new(**context)
+
+      expect(source.send(:summary_select_fields)).to eq(
+        [
+          'dimension_id',
+          "SUM(CASE WHEN metric = 'resolutions_count' THEN count ELSE 0 END) as resolved_conversations_count",
+          "SUM(CASE WHEN metric = 'resolution_time' THEN count ELSE 0 END) as avg_resolution_time_count",
+          "SUM(CASE WHEN metric = 'resolution_time' THEN sum_value ELSE 0 END) as avg_resolution_time_sum_value",
+          "SUM(CASE WHEN metric = 'first_response' THEN count ELSE 0 END) as avg_first_response_time_count",
+          "SUM(CASE WHEN metric = 'first_response' THEN sum_value ELSE 0 END) as avg_first_response_time_sum_value",
+          "SUM(CASE WHEN metric = 'reply_time' THEN count ELSE 0 END) as avg_reply_time_count",
+          "SUM(CASE WHEN metric = 'reply_time' THEN sum_value ELSE 0 END) as avg_reply_time_sum_value"
+        ]
+      )
+    end
+  end
+
+  describe 'adapter contract' 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 => {
+              conversations_count: 2,
+              resolved_conversations_count: 2,
+              avg_resolution_time: 150.0,
+              avg_first_response_time: 30.0,
+              avg_reply_time: 20.0
+            },
+            inbox2.id => {
+              conversations_count: 1,
+              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 => {
+                conversations_count: 2,
+                resolved_conversations_count: 2,
+                avg_resolution_time: 115.0,
+                avg_first_response_time: 20.0,
+                avg_reply_time: 12.5
+              },
+              inbox2.id => {
+                conversations_count: 1,
+                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
+end