feat: extract raw report queries into DataSource abstraction

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

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