diff --git a/app/services/reports/data_source.rb b/app/services/reports/data_source.rb new file mode 100644 index 000000000..7c013295d --- /dev/null +++ b/app/services/reports/data_source.rb @@ -0,0 +1,106 @@ +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], context[:timezone_offset]) && + supported_metric?(context[:metric]) + end + + def timezone_matches_account?(account, timezone, timezone_offset) + return normalized_timezone_identifier(timezone) == normalized_timezone_identifier(account.reporting_timezone) if timezone.present? + + 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 + + def normalized_timezone_identifier(timezone) + ActiveSupport::TimeZone[timezone]&.tzinfo&.name + 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 summary_metrics + @summary_metrics ||= ReportingEvents::MetricRegistry.summary_metrics + 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..5abcd7176 --- /dev/null +++ b/app/services/reports/raw_data_source.rb @@ -0,0 +1,156 @@ +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 + metric_results = summary_scope + .select(*summary_select_fields) + .group(summary_group_by_key) + .index_by { |record| record.public_send(summary_index_key) } + + merge_summary_results(metric_results, summary_conversation_counts) + 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_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).each_with_object({}) do |dimension_id, results| + record = metric_results[dimension_id] + results[dimension_id] = summary_attributes_for(record, conversation_counts[dimension_id]) + end + end + + def summary_select_fields + ["#{summary_group_by_key} as #{summary_index_key}"] + summary_metrics.map { |definition| summary_select_field(definition) } + end + + def summary_select_field(definition) + if definition[:aggregate] == :count + "COUNT(CASE WHEN name = '#{definition[:raw_event_name]}' THEN 1 END) as #{definition[:summary_key]}" + else + "AVG(CASE WHEN name = '#{definition[:raw_event_name]}' THEN #{average_value_key} END) as #{definition[:summary_key]}" + end + end + + def summary_attributes_for(record, conversations_count = 0) + summary_metrics.each_with_object({ conversations_count: conversations_count.to_i }) do |definition, attributes| + value = record&.public_send(definition[:summary_key]) + attributes[definition[:summary_key]] = definition[:aggregate] == :count ? value.to_i : value + end + end + + def summary_group_by_key + { + 'account' => :account_id, + 'agent' => :user_id, + 'inbox' => :inbox_id, + 'team' => 'conversations.team_id' + }[dimension_type] + end + + def summary_conversation_group_by_key + { + 'account' => :account_id, + 'agent' => :assignee_id, + 'inbox' => :inbox_id, + 'team' => :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