feat(catalog): wire metered billing through ProcessService

app/models/billing_segment.rb

@@ -45,6 +45,10 @@ def rate
     rate_override || rate_card_rate
   end
 
+  def target_key
+    "contract-#{contract_id}-#{contract_rate_card.product.target_key}"
+  end
+
   # The shared matcher expects a filter object for the default bucket, not nil.
   # This mirrors the empty ChargeFilter in app/services/fees/charge_service.rb:85-95.
   def empty_product_filter

app/services/billing_segments/process_service.rb

@@ -36,8 +36,13 @@ def call
     def pending_segments
       BillingSegment.status_pending
         .where(customer_id: customer.id)
+        .joins(contract_rate_card: {rate_card: :product})
+        .where(
+          "products.product_type != :metered OR rate_cards.billing_timing != :advance",
+          metered: Product::PRODUCT_TYPES[:metered],
+          advance: RateCard::BILLING_TIMINGS[:advance]
+        )
         .includes(:pricing_unit, :rate_override, :contract, contract_rate_card: {rate_card: :product}, rate_card_rate: :rate_card)
-        .select { |segment| segment.contract_rate_card.product.fixed? }
     end
 
     def invoice_key(segment)
@@ -67,6 +72,9 @@ def payment_method_key(contract)
     def build_invoice(segments)
       contract = segments.first.contract
       invoice = nil
+      grouped = segments.group_by { |s| s.contract_rate_card.product.product_type }
+      metered_segments = grouped[Product::PRODUCT_TYPES[:metered]] || []
+      fixed_segments = grouped[Product::PRODUCT_TYPES[:fixed]] || []
 
       ActiveRecord::Base.transaction do
         invoice = Invoices::CreateGeneratingService.call!(
@@ -79,17 +87,11 @@ def build_invoice(segments)
           purchase_order_number: contract.purchase_order_number
         ).invoice
 
-        segments.flat_map do |segment|
-          if segment.contract_rate_card.product.fixed?
-            computed_fees(segment)
-          else
-            []
-          end
-        end.each do |fee|
-          fee.invoice = invoice
-          fee.billing_entity = invoice.billing_entity
-          fee.save!
-        end
+        filtered_aggregations = event_filters(metered_segments)
+
+        attach_fixed_fees(fixed_segments, invoice)
+        attach_metered_fees(metered_segments, invoice, filtered_aggregations)
+
         invoice.fees.reload
 
         Invoices::ComputeAmountsFromFees.call!(invoice:)
@@ -100,11 +102,47 @@ def build_invoice(segments)
       invoice
     end
 
-    def computed_fees(segment)
+    def attach_fixed_fees(segments, invoice)
+      segments.each do |segment|
+        compute_fixed_fees(segment).each do |fee|
+          fee.invoice = invoice
+          fee.billing_entity = invoice.billing_entity
+          fee.save!
+        end
+      end
+    end
+
+    def attach_metered_fees(segments, invoice, filtered_aggregations)
+      segments.each do |segment|
+        compute_metered_fees(segment, invoice, filtered_aggregations)
+      end
+    end
+
+    def compute_fixed_fees(segment)
       fee_result = BillingSegments::Fees::ComputeService.call!(billing_segment: segment)
       [fee_result.fee, fee_result.true_up_fee].compact
     end
 
+    # NOTE: Fees::ChargeService persists and attaches the product fees itself (within this
+    # surrounding transaction), so a failure on any segment rolls back the whole invoice group.
+    def compute_metered_fees(segment, invoice, filtered_aggregations)
+      ::Fees::ChargeService.call!(
+        invoice:,
+        metered_item: ::Fees::ChargeService::MeteredItem.from_billing_segment(segment),
+        billing_context: Billing::Context.from(contract: segment.contract),
+        options: ::Fees::ChargeService::Options.new(context: :finalize, skip_adjusted_fees: true),
+        filtered_aggregations: filtered_aggregations[segment.target_key]&.keys || []
+      )
+    end
+
+    def event_filters(metered_segments)
+      return {} if metered_segments.empty?
+
+      Events::BillingPeriodFilterService.for_billing_segments!(
+        billing_segments: metered_segments, with_last_seen_at: false
+      ).filter_targets
+    end
+
     def finalize_generating_invoices
       # Re-query done segments so a retry finalizes existing invoices without rebuilding fees.
       invoice_ids = BillingSegment.status_done

app/services/events/billing_period_filter_service.rb

@@ -15,10 +15,9 @@ def self.for_charges!(subscription:, boundaries:, codes: nil, with_last_seen_at:
       )
     end
 
-    def self.for_billing_segments!(contract:, billing_segments:, codes: nil, with_last_seen_at: true)
+    def self.for_billing_segments!(billing_segments:, codes: nil, with_last_seen_at: true)
       call!(
         resolver: BillingPeriodFilters::BillingSegmentsResolver.new(
-          contract:,
           billing_segments:,
           codes:,
           with_last_seen_at:

app/services/events/billing_period_filters/base_resolver.rb

@@ -30,7 +30,7 @@ def event_values_with_history
       end
 
       def non_recurring_metric_codes
-        @non_recurring_metric_codes ||= metric_codes.to_a - recurring_metric_codes
+        @non_recurring_metric_codes ||= metric_codes(record_id: nil).to_a - recurring_metric_codes
       end
 
       def recurring_event_filter_targets

app/services/events/billing_period_filters/billing_segments_resolver.rb

@@ -3,41 +3,66 @@
 module Events
   module BillingPeriodFilters
     class BillingSegmentsResolver < BaseResolver
-      def initialize(contract:, billing_segments:, codes: nil, with_last_seen_at: true)
-        @contract = contract
+      def initialize(billing_segments:, codes: nil, with_last_seen_at: true)
         @billing_segments = billing_segments
         @codes = codes&.to_set || Set.new
         @with_last_seen_at = with_last_seen_at
       end
 
       def filter_targets
         return {} if target_segments.empty?
-        return {} if metric_codes.empty?
+        return {} if metric_codes_by_contract_id.empty?
+
+        # Event stores are scoped to a contract's external_id, so combinations must only
+        # be matched against targets from that contract, even when products share a code.
+        # For example, Contract A and Contract B can both use product "api_calls", while
+        # only A has a {region: "eu"} event. Matching A's combinations against B would
+        # incorrectly add A's filter bucket and last_seen_at to B's target.
+        result = recurring_event_filter_targets
+
+        contracts.each do |contract|
+          combinations = event_values_with_history_for(contract_id: contract.id) do |**options|
+            event_store_for(contract).distinct_codes_and_property_combinations(
+              filter_keys: billable_metric_filter_keys_by_contract.fetch(contract.id, []),
+              **options
+            )
+          end
+
+          event_codes = combinations.map(&:first).to_set
+          contract_targets = segments_by_contract.fetch(contract.id, []).select do |target|
+            event_codes.include?(filter_target_for(target).billable_metric.code)
+          end
+
+          filter_targets_from_combinations(
+            combinations:,
+            targets: contract_targets,
+            result:
+          )
+        end
 
-        super
+        result
       end
 
       private
 
-      attr_reader :contract, :billing_segments, :codes, :with_last_seen_at
+      attr_reader :billing_segments, :codes, :with_last_seen_at
 
-      delegate :organization, to: :contract
+      def organization
+        @organization ||= target_segments.first.organization
+      end
 
       def filter_target_for(billing_segment)
-        Events::BillingPeriodFilters::FilterTarget.from_billing_segment(billing_segment:)
+        @filter_targets ||= {}
+        @filter_targets[billing_segment.id] ||= Events::BillingPeriodFilters::FilterTarget.from_billing_segment(billing_segment:)
       end
 
       def target_segments
         @target_segments ||= billing_segments_scope.preload(
+          :contract,
           contract_rate_card: {product: [:billable_metric, {filters: {values: :billable_metric_filter}}]}
         ).to_a
       end
 
-      def targets_with_events(codes)
-        event_codes = codes.to_set
-        target_segments.select { |segment| event_codes.include?(filter_target_for(segment).billable_metric.code) }
-      end
-
       def billing_segments_scope
         scope = BillingSegment.where(id: billing_segments)
           .joins(contract_rate_card: {product: :billable_metric})
@@ -49,15 +74,38 @@ def billing_segments_scope
         end
       end
 
-      def metric_codes
-        @metric_codes ||= codes.presence || billing_segments_scope.distinct.pluck("billable_metrics.code")
+      def event_values_with_history_for(contract_id:)
+        recurring_codes = recurring_metric_codes_by_contract_id.fetch(contract_id, [])
+        non_recurring_metric_codes = metric_codes_for(contract_id:) - recurring_codes
+
+        values = yield(codes: non_recurring_metric_codes, with_last_seen_at:)
+
+        if recurring_codes.any?
+          values += yield(codes: recurring_codes, include_all_history: true, with_last_seen_at:)
+        end
+
+        values
+      end
+
+      def metric_codes_for(contract_id:)
+        @metric_codes_by_contract ||= {}
+        @metric_codes_by_contract[contract_id] ||= codes.presence || metric_codes_by_contract_id.fetch(contract_id, [])
       end
 
-      def recurring_metric_codes
-        @recurring_metric_codes ||= billing_segments_scope
+      def metric_codes_by_contract_id
+        @metric_codes_by_contract_id ||= billing_segments_scope
+          .distinct
+          .pluck("contract_rate_cards.contract_id, billable_metrics.code")
+          .group_by(&:first).transform_values! { |pairs| pairs.map(&:last) }
+      end
+
+      def recurring_metric_codes_by_contract_id
+        @recurring_metric_codes_by_contract_id ||= billing_segments_scope
           .where(billable_metrics: {recurring: true})
           .distinct
-          .pluck("billable_metrics.code")
+          .pluck("contract_rate_cards.contract_id, billable_metrics.code")
+          .group_by(&:first)
+          .transform_values! { |pairs| pairs.map(&:last) }
       end
 
       def current_recurring_targets
@@ -68,22 +116,36 @@ def period_start
         @period_start ||= target_segments.map(&:started_at).min
       end
 
-      def billable_metric_filter_keys
-        @billable_metric_filter_keys ||= billing_segments_scope
+      def billable_metric_filter_keys_by_contract
+        @billable_metric_filter_keys_by_contract ||= billing_segments_scope
           .joins(contract_rate_card: {product: {billable_metric: :filters}})
           .distinct
-          .pluck("billable_metric_filters.key")
-      end
-
-      def event_store
-        @event_store ||= Events::Stores::StoreFactory.new_instance(
-          organization:,
-          billing_context: Billing::Context.from(contract:),
-          boundaries: {
-            from_datetime: period_start,
-            to_datetime: target_segments.map(&:ended_at).max
-          }
-        )
+          .pluck("contract_rate_cards.contract_id", "billable_metric_filters.key")
+          .group_by(&:first)
+          .transform_values! { |pairs| pairs.map(&:last) }
+      end
+
+      def event_store_for(contract)
+        @event_stores ||= {}
+        @event_stores[contract.id] ||= begin
+          contract_segments = segments_by_contract[contract.id]
+          Events::Stores::StoreFactory.new_instance(
+            organization:,
+            billing_context: Billing::Context.from(contract:),
+            boundaries: {
+              from_datetime: contract_segments.map(&:started_at).min,
+              to_datetime: contract_segments.map(&:ended_at).max
+            }
+          )
+        end
+      end
+
+      def segments_by_contract
+        @segments_by_contract ||= target_segments.group_by(&:contract_id)
+      end
+
+      def contracts
+        @contracts ||= target_segments.map(&:contract).uniq
       end
     end
   end

app/services/events/billing_period_filters/charges_resolver.rb

@@ -33,12 +33,13 @@ def event_store
 
       # A code outside of the plan matches no event, so codes is used as is: dropping it would leave
       # its charge out of the result, billed as zero units instead of surfaced.
-      def metric_codes
+      def metric_codes(record_id: nil)
         @metric_codes ||= codes || plan.billable_metrics.distinct.pluck(:code)
       end
 
       def filter_target_for(charge)
-        Events::BillingPeriodFilters::FilterTarget.from_charge(charge:)
+        @filter_target_for ||= {}
+        @filter_target_for[charge.id] ||= Events::BillingPeriodFilters::FilterTarget.from_charge(charge:)
       end
 
       def targets_with_events(codes)

app/services/events/billing_period_filters/sources/billing_segment.rb

@@ -32,7 +32,7 @@ def all_filter_values?(filter, key)
           filter.to_h[key] == [nil]
         end
 
-        delegate :target_key, to: :product
+        delegate :target_key, to: :billing_segment
 
         def with_filter(filter)
           self.class.new(billing_segment:, filter:)

spec/models/billing_segment_spec.rb

@@ -110,6 +110,32 @@
     end
   end
 
+  describe "#target_key" do
+    let(:organization) { create(:organization) }
+    let(:customer) { create(:customer, organization:) }
+    let(:contract) { create(:contract, organization:, customer:) }
+    let(:product) { create(:product, organization:) }
+    let(:rate_card) { create(:rate_card, organization:, product:) }
+    let(:contract_rate_card) { create(:contract_rate_card, organization:, contract:, rate_card:) }
+    let(:rate_card_rate) { create(:rate_card_rate, organization:, rate_card:) }
+    let(:billing_segment) do
+      create(
+        :billing_segment,
+        organization:,
+        customer:,
+        contract:,
+        contract_rate_card:,
+        rate_card_rate:
+      )
+    end
+
+    it "includes the contract and product identities" do
+      expect(billing_segment.target_key).to eq(
+        "contract-#{billing_segment.contract_id}-#{billing_segment.contract_rate_card.product.target_key}"
+      )
+    end
+  end
+
   describe "#duration_in_days" do
     subject(:segment) { described_class.new(customer:, started_at:, ended_at:) }