refactor(events): pass event contexts

This change unblocks the initial BillingSegment refactor needed to support BillingPeriodFilterService. BillingSegment filtering queries raw events through contract data, while the existing event stores and billable metric aggregators assumed every event context was a Subscription.

Introduce Events::Stores::EventContext as the explicit object passed into event stores and aggregators. The stores now ask the context for external ids, organization, customer, duration, subscription history, and date calculations instead of depending directly on a subscription.

Contract-backed contexts currently support the raw event lookup surface needed by BillingSegment, and intentionally raise NotImplementedError for subscription-specific methods that are not defined yet: invoice_subscriptions, previous_subscription, previous_subscription_id, previous_subscription_id?, next_subscription, upgraded?, downgraded?, id, and charges_duration_at. Those contract behaviours will be addressed in a separate PR.

Author: tiagolupepic

app/services/billable_metrics/aggregation_factory.rb

@@ -2,10 +2,11 @@
 
 module BillableMetrics
   class AggregationFactory
-    def self.new_instance(charge:, current_usage: false, **attributes)
+    def self.new_instance(charge:, context:, current_usage: false, **attributes)
       aggregator_class(charge, current_usage).new(
         event_store_class: Events::Stores::StoreFactory.store_class(organization: charge.billable_metric.organization),
         charge:,
+        context:,
         **attributes
       )
     end

app/services/billable_metrics/aggregations/base_service.rb

@@ -48,11 +48,11 @@ def self.null_result(result, grouped_by_keys: nil, apply_aggregation: false)
         result
       end
 
-      def initialize(event_store_class:, charge:, subscription:, boundaries:, filters: {}, bypass_aggregation: false)
+      def initialize(event_store_class:, charge:, context:, boundaries:, filters: {}, bypass_aggregation: false)
         super(nil)
         @event_store_class = event_store_class
         @charge = charge
-        @subscription = subscription
+        @context = context
 
         @filters = filters
         @charge_filter = filters[:charge_filter]
@@ -133,7 +133,7 @@ def empty_results
 
       attr_accessor :event_store_class,
         :charge,
-        :subscription,
+        :context,
         :filters,
         :charge_filter,
         :event,
@@ -146,12 +146,12 @@ def empty_results
 
       delegate :billable_metric, to: :charge
 
-      delegate :customer, to: :subscription
+      delegate :customer, to: :context
 
       def event_store
         @event_store ||= event_store_class.new(
           code: billable_metric.code,
-          subscription:,
+          context:,
           boundaries:,
           filters:,
           deduplicate: deduplicate?
@@ -162,7 +162,7 @@ def deduplicate?
         override = Events::Stores::StoreFactory.override
         return override[:deduplicate] if override
 
-        organization = subscription&.organization
+        organization = context&.organization
         return false unless organization
 
         organization.clickhouse_events_store? && organization.clickhouse_deduplication_enabled?
@@ -230,7 +230,7 @@ def empty_result
       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(external_subscription_id: context.external_id)
           .where(charge_id: charge.id)
           .from_datetime(with_from_datetime)
           .to_datetime(with_to_datetime)

app/services/billable_metrics/aggregations/custom_service.rb

@@ -96,7 +96,7 @@ def latest_state(grouped_by_values:)
 
         query = CachedAggregation
           .where(organization_id: billable_metric.organization_id)
-          .where(external_subscription_id: subscription.external_id)
+          .where(external_subscription_id: context.external_id)
           .where(charge_id: charge.id)
           .where("cached_aggregations.timestamp < ?", truncated_datetime)
           .where(grouped_by: grouped_by_values.presence || {})
@@ -128,7 +128,7 @@ def perform_custom_aggregation(target_result: result, grouped_by_values: nil)
         if grouped_by_values
           store = event_store_class.new(
             code: billable_metric.code,
-            subscription:,
+            context:,
             boundaries:,
             filters: filters.merge(grouped_by_values:)
           )

app/services/billable_metrics/aggregations/weighted_sum_service.rb

@@ -101,7 +101,7 @@ def latest_value
           return @latest_value = latest_cached_aggregation.current_aggregation
         end
 
-        if subscription.previous_subscription_id?
+        if context.previous_subscription_id?
           return @latest_value = latest_value_from_events.first
         end
 
@@ -115,7 +115,7 @@ def latest_breakdowns
           return @latest_breakdowns = latest_cached_aggregation.presentation_breakdowns
         end
 
-        if subscription.previous_subscription_id?
+        if context.previous_subscription_id?
           return @latest_breakdowns = latest_value_from_events.second
         end
 
@@ -135,7 +135,7 @@ def latest_cached_aggregations
 
         query = CachedAggregation
           .where(organization_id: billable_metric.organization_id)
-          .where(external_subscription_id: subscription.external_id)
+          .where(external_subscription_id: context.external_id)
           .where(charge_id: charge.id)
           .where(timestamp: ...from_datetime)
           .order(timestamp: :desc, created_at: :desc)
@@ -152,7 +152,7 @@ def latest_value_from_events
 
         event_store = event_store_class.new(
           code: billable_metric.code,
-          subscription:,
+          context:,
           boundaries: {to_datetime: from_datetime - 1.second},
           filters:
         )
@@ -178,7 +178,7 @@ def grouped_latest_values
           end
         end
 
-        if subscription.previous_subscription_id?
+        if context.previous_subscription_id?
           return @grouped_latest_values = grouped_latest_values_from_events.first
         end
 
@@ -192,7 +192,7 @@ def grouped_latest_breakdowns
           return @grouped_latest_breakdowns = grouped_latest_cached_aggregations.flat_map(&:presentation_breakdowns)
         end
 
-        if subscription.previous_subscription_id?
+        if context.previous_subscription_id?
           return @grouped_latest_breakdowns = grouped_latest_values_from_events.second
         end
 
@@ -214,7 +214,7 @@ def grouped_latest_cached_aggregations
       def grouped_latest_values_from_events
         event_store = event_store_class.new(
           code: billable_metric.code,
-          subscription:,
+          context:,
           boundaries: {to_datetime: from_datetime - 1.second},
           filters:
         )

app/services/billable_metrics/breakdown/sum_service.rb

@@ -27,7 +27,7 @@ def to_date_in_customer_timezone
       def persisted_breakdown
         event_store = event_store_class.new(
           code: billable_metric.code,
-          subscription:,
+          context:,
           boundaries: {to_datetime: from_datetime},
           filters:
         )

app/services/billable_metrics/prorated_aggregations/base_service.rb

@@ -101,7 +101,7 @@ def period_duration
       #       we want to bill the persisted metrics at prorata of the full period duration.
       #       ie: the number of day of the terminated period divided by number of days without termination
       def persisted_pro_rata
-        subscription.date_diff_with_timezone(from_datetime, to_datetime).fdiv(period_duration)
+        context.date_diff_with_timezone(from_datetime, to_datetime).fdiv(period_duration)
       end
 
       attr_accessor :options

app/services/billable_metrics/prorated_aggregations/sum_service.rb

@@ -148,7 +148,7 @@ def persisted_event_store_instance
         @persisted_event_store_instance ||= begin
           event_store = event_store_class.new(
             code: billable_metric.code,
-            subscription:,
+            context:,
             boundaries: {to_datetime: from_datetime - PERSISTED_TOP_BOUNDARY_DELAY}, # Note: Avoid counting events exactly on `from_datetime` twice
             filters:,
             deduplicate: deduplicate?
@@ -175,7 +175,7 @@ def current_prorated_result
       def persisted_prorated_result
         @persisted_prorated_result ||= persisted_event_store_instance.prorated_sum(
           period_duration:,
-          persisted_duration: subscription.date_diff_with_timezone(from_datetime, to_datetime)
+          persisted_duration: context.date_diff_with_timezone(from_datetime, to_datetime)
         )
       end
 
@@ -242,7 +242,7 @@ def current_grouped_prorated_results
       def persisted_grouped_prorated_results
         @persisted_grouped_prorated_results ||= persisted_event_store_instance.grouped_prorated_sum(
           period_duration:,
-          persisted_duration: subscription.date_diff_with_timezone(from_datetime, to_datetime)
+          persisted_duration: context.date_diff_with_timezone(from_datetime, to_datetime)
         )
       end
     end

app/services/charges/pay_in_advance_aggregation_service.rb

@@ -17,7 +17,7 @@ def initialize(charge:, boundaries:, properties:, event:, charge_filter: nil)
     def call
       aggregator = BillableMetrics::AggregationFactory.new_instance(
         charge:,
-        subscription:,
+        context: Events::Stores::EventContext.from(subscription:),
         boundaries: {
           from_datetime: boundaries.charges_from_datetime,
           to_datetime: boundaries.charges_to_datetime,

app/services/events/billing_period_filters/charges_resolver.rb

@@ -31,7 +31,7 @@ def period_start
       def event_store
         @event_store ||= Events::Stores::StoreFactory.new_instance(
           organization: organization,
-          subscription:,
+          context: Events::Stores::EventContext.from(subscription:),
           boundaries: {
             from_datetime: boundaries.charges_from_datetime,
             to_datetime: boundaries.charges_to_datetime

app/services/events/stores/base_store.rb

@@ -56,9 +56,9 @@ def to_grouped_hash
         :events_count
       )
 
-      def initialize(subscription:, boundaries:, code: nil, filters: {}, deduplicate: false)
+      def initialize(context:, boundaries:, code: nil, filters: {}, deduplicate: false)
         @code = code
-        @subscription = subscription
+        @context = context
         @boundaries = boundaries
 
         @filters = filters
@@ -183,7 +183,7 @@ def to_datetime
       end
 
       def charges_duration
-        boundaries[:charges_duration]
+        period_duration
       end
 
       def applicable_to_datetime
@@ -213,16 +213,12 @@ def with_presentation_by_in_grouped_by?
 
       protected
 
-      attr_accessor :code, :subscription, :boundaries, :grouped_by_values, :filters, :matching_filters, :ignored_filters, :deduplicate
+      attr_accessor :code, :context, :boundaries, :grouped_by_values, :filters, :matching_filters, :ignored_filters, :deduplicate
 
-      delegate :customer, to: :subscription
+      delegate :customer, to: :context
 
       def period_duration
-        @period_duration ||= Subscriptions::DatesService.new_instance(
-          subscription,
-          to_datetime + 1.day,
-          current_usage: subscription.terminated? && subscription.upgraded?
-        ).charges_duration_in_days
+        @period_duration ||= context.charges_duration_at(to_datetime + 1.day)
       end
 
       def build_aggregation_result(row)