From 692b05766b79de365d7e6f125d7084d3323abc83 Mon Sep 17 00:00:00 2001 From: Shivam Mishra Date: Wed, 11 Mar 2026 18:58:52 +0530 Subject: [PATCH] refactor: add report data sources --- app/services/reports/data_source.rb | 96 +++++++ app/services/reports/raw_data_source.rb | 143 ++++++++++ app/services/reports/rollup_data_source.rb | 170 +++++++++++ .../data_source_adapter_contract_spec.rb | 268 ++++++++++++++++++ spec/services/reports/data_source_spec.rb | 62 ++++ 5 files changed, 739 insertions(+) create mode 100644 app/services/reports/data_source.rb create mode 100644 app/services/reports/raw_data_source.rb create mode 100644 app/services/reports/rollup_data_source.rb create mode 100644 spec/services/reports/data_source_adapter_contract_spec.rb create mode 100644 spec/services/reports/data_source_spec.rb diff --git a/app/services/reports/data_source.rb b/app/services/reports/data_source.rb new file mode 100644 index 000000000..d93bed875 --- /dev/null +++ b/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 diff --git a/app/services/reports/raw_data_source.rb b/app/services/reports/raw_data_source.rb new file mode 100644 index 000000000..07be03097 --- /dev/null +++ b/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 diff --git a/app/services/reports/rollup_data_source.rb b/app/services/reports/rollup_data_source.rb new file mode 100644 index 000000000..1f2383bfa --- /dev/null +++ b/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 diff --git a/spec/services/reports/data_source_adapter_contract_spec.rb b/spec/services/reports/data_source_adapter_contract_spec.rb new file mode 100644 index 000000000..33e1ffcd9 --- /dev/null +++ b/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 diff --git a/spec/services/reports/data_source_spec.rb b/spec/services/reports/data_source_spec.rb new file mode 100644 index 000000000..fb7bd2e3b --- /dev/null +++ b/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