feat: add rollup data source and route eligible reports through it
This commit is contained in:
Shivam Mishra
@@ -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
|
||||
@@ -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
|
||||
Reference in New Issue
Block a user