feat(grouped-by): Cleanup and fixes to finalize the feature

app/models/event.rb

@@ -40,7 +40,11 @@ def customer
 
   def subscription
     scope = if external_customer_id && customer
-      customer.subscriptions
+      if external_subscription_id
+        customer.subscriptions.where(external_id: external_subscription_id)
+      else
+        customer.subscriptions
+      end
     else
       organization.subscriptions.where(external_id: external_subscription_id)
     end

app/services/billable_metrics/aggregations/base_service.rb

@@ -76,7 +76,11 @@ def to_datetime
       end
 
       def handle_in_advance_current_usage(total_aggregation, target_result: result)
-        cached_aggregation = find_cached_aggregation(grouped_by: target_result.grouped_by)
+        cached_aggregation = find_cached_aggregation(
+          with_from_datetime: from_datetime,
+          with_to_datetime: to_datetime,
+          grouped_by: target_result.grouped_by,
+        )
 
         if cached_aggregation
           aggregation = total_aggregation -

app/services/billable_metrics/aggregations/sum_service.rb

@@ -95,7 +95,11 @@ def compute_pay_in_advance_aggregation
 
         value = event.properties.fetch(billable_metric.field_name, 0).to_s
 
-        cached_aggregation = find_cached_aggregation
+        cached_aggregation = find_cached_aggregation(
+          with_from_datetime: from_datetime,
+          with_to_datetime: to_datetime,
+          grouped_by: grouped_by_values,
+        )
 
         unless cached_aggregation
           return_value = BigDecimal(value).negative? ? '0' : value
@@ -129,21 +133,16 @@ def compute_per_event_aggregation
       # This method fetches the latest cached aggregation in current period. If such a record exists we know that
       # previous aggregation and previous maximum aggregation are stored there. Fetching these values
       # would help us in pay in advance value calculation without iterating through all events in current period
-      def find_cached_aggregation(with_from_datetime: from_datetime, with_to_datetime: to_datetime, grouped_by: nil)
+      def find_cached_aggregation(with_from_datetime:, with_to_datetime:, grouped_by: nil)
         query = CachedAggregation
           .where(organization_id: billable_metric.organization_id)
           .where(external_subscription_id: subscription.external_id)
           .where(charge_id: charge.id)
           .from_datetime(with_from_datetime)
           .to_datetime(with_to_datetime)
+          .where(grouped_by: grouped_by.presence || {})
           .order(timestamp: :desc)
 
-        query = if grouped_by.present?
-          query.where(grouped_by:)
-        else
-          query.where(grouped_by: {})
-        end
-
         query = query.where.not(event_id: event.id) if event.present?
         query = query.where(group_id: group.id) if group
 

app/services/billable_metrics/aggregations/unique_count_service.rb

@@ -59,7 +59,11 @@ def compute_pay_in_advance_aggregation
 
         newly_applied_units = (operation_type == :add) ? 1 : 0
 
-        cached_aggregation = find_cached_aggregation(grouped_by: grouped_by_values)
+        cached_aggregation = find_cached_aggregation(
+          with_from_datetime: from_datetime,
+          with_to_datetime: to_datetime,
+          grouped_by: grouped_by_values,
+        )
 
         unless cached_aggregation
           handle_event_metadata(
@@ -105,7 +109,7 @@ def compute_per_event_aggregation
       # This method fetches the latest cached aggregation in current period. If such a record exists we know that
       # previous aggregation and previous maximum aggregation are stored there. Fetching these values
       # would help us in pay in advance value calculation without iterating through all events in current period
-      def find_cached_aggregation(with_from_datetime: from_datetime, with_to_datetime: to_datetime, grouped_by: nil)
+      def find_cached_aggregation(with_from_datetime:, with_to_datetime:, grouped_by: nil)
         query = CachedAggregation
           .where(organization_id: billable_metric.organization_id)
           .where(external_subscription_id: subscription.external_id)
@@ -141,7 +145,7 @@ def find_cached_aggregation(with_from_datetime: from_datetime, with_to_datetime:
             .where('quantified_events.group_id = cached_aggregations.group_id')
         end
 
-        @cached_aggregation = query.first
+        query.first
       end
 
       def count_unique_group_scope(events)

app/services/billable_metrics/prorated_aggregations/base_service.rb

@@ -3,11 +3,7 @@
 module BillableMetrics
   module ProratedAggregations
     class BaseService < BillableMetrics::Aggregations::BaseService
-      def aggregation_without_proration
-        @aggregation_without_proration ||= base_aggregator.aggregate(options:)
-      end
-
-      def compute_pay_in_advance_aggregation
+      def compute_pay_in_advance_aggregation(aggregation_without_proration:)
         return BigDecimal(0) unless event
         return BigDecimal(0) if event.properties.blank?
 
@@ -25,14 +21,14 @@ def compute_pay_in_advance_aggregation
 
         value = (result_without_proration * proration_coefficient).ceil(5)
 
-        extend_cached_aggregation(value)
+        extend_cached_aggregation(value, aggregation_without_proration)
 
         value
       end
 
       # We need to extend cached aggregation with max_aggregation_with_proration. This attribute will be used
       # for current usage in pay_in_advance case
-      def extend_cached_aggregation(prorated_value)
+      def extend_cached_aggregation(prorated_value, aggregation_without_proration)
         result.max_aggregation = aggregation_without_proration.max_aggregation
         result.current_aggregation = aggregation_without_proration.current_aggregation
 
@@ -60,11 +56,7 @@ def extend_cached_aggregation(prorated_value)
       # In current usage section two main values are presented, number of units in period and amount.
       # Proration affects only amount (calculated from aggregation) and number of units shows full number of units
       # (calculated from current_usage_units).
-      def handle_current_usage(
-        result_with_proration, is_pay_in_advance, target_result: result,
-        aggregation_without_proration: nil
-      )
-        aggregation_without_proration ||= self.aggregation_without_proration
+      def handle_current_usage(result_with_proration, is_pay_in_advance, aggregation_without_proration:, target_result:)
         value_without_proration = aggregation_without_proration.aggregation
         cached_aggregation = base_aggregator.find_cached_aggregation(
           with_from_datetime: from_datetime,

app/services/billable_metrics/prorated_aggregations/sum_service.rb

@@ -13,7 +13,7 @@ def initialize(**args)
       end
 
       def compute_aggregation(options: {})
-        @options = options
+        aggregation_without_proration = base_aggregator.aggregate(options:)
 
         # For charges that are pay in advance on billing date we always bill full amount
         return aggregation_without_proration if event.nil? && options[:is_pay_in_advance] && !options[:is_current_usage]
@@ -22,12 +22,17 @@ def compute_aggregation(options: {})
         result.full_units_number = aggregation_without_proration.aggregation if event.nil?
 
         if options[:is_current_usage]
-          handle_current_usage(aggregation, options[:is_pay_in_advance])
+          handle_current_usage(
+            aggregation,
+            options[:is_pay_in_advance],
+            target_result: result,
+            aggregation_without_proration:,
+          )
         else
           result.aggregation = aggregation
         end
 
-        result.pay_in_advance_aggregation = compute_pay_in_advance_aggregation
+        result.pay_in_advance_aggregation = compute_pay_in_advance_aggregation(aggregation_without_proration:)
         result.count = aggregation_without_proration.count
         result.options = options
         result
@@ -44,7 +49,7 @@ def compute_aggregation(options: {})
       #       as pay in advance aggregation will be computed on a single group
       #       with the grouped_by_values filter
       def compute_grouped_by_aggregation(options: {})
-        @options = options
+        aggregation_without_proration = base_aggregator.aggregate(options:)
 
         # For charges that are pay in advance on billing date we always bill full amount
         return aggregation_without_proration if event.nil? && options[:is_pay_in_advance] && !options[:is_current_usage]
@@ -59,6 +64,11 @@ def compute_grouped_by_aggregation(options: {})
             agg.grouped_by == aggregation[:groups]
           end
 
+          unless group_result_without_proration
+            group_result_without_proration = empty_results.aggregations.first
+            group_result_without_proration.grouped_by = aggregation[:groups]
+          end
+
           group_result = BaseService::Result.new
           group_result.grouped_by = aggregation[:groups]
           group_result.full_units_number = group_result_without_proration&.aggregation || 0

app/services/billable_metrics/prorated_aggregations/unique_count_service.rb

@@ -10,7 +10,7 @@ def initialize(**args)
       end
 
       def compute_aggregation(options: {})
-        @options = options
+        aggregation_without_proration = base_aggregator.aggregate(options:)
 
         # For charges that are pay in advance on billing date we always bill full amount
         return aggregation_without_proration if event.nil? && options[:is_pay_in_advance] && !options[:is_current_usage]
@@ -19,12 +19,17 @@ def compute_aggregation(options: {})
         result.full_units_number = aggregation_without_proration.aggregation if event.nil?
 
         if options[:is_current_usage]
-          handle_current_usage(aggregation, options[:is_pay_in_advance])
+          handle_current_usage(
+            aggregation,
+            options[:is_pay_in_advance],
+            target_result: result,
+            aggregation_without_proration:,
+          )
         else
           result.aggregation = aggregation
         end
 
-        result.pay_in_advance_aggregation = compute_pay_in_advance_aggregation
+        result.pay_in_advance_aggregation = compute_pay_in_advance_aggregation(aggregation_without_proration:)
         result.options = options
         result.count = result.aggregation
         result
@@ -39,7 +44,7 @@ def compute_aggregation(options: {})
       #       as pay in advance aggregation will be computed on a single group
       #       with the grouped_by_values filter
       def compute_grouped_by_aggregation(options: {})
-        @options = options
+        aggregation_without_proration = base_aggregator.aggregate(options:)
 
         # For charges that are pay in advance on billing date we always bill full amount
         return aggregation_without_proration if event.nil? && options[:is_pay_in_advance] && !options[:is_current_usage]