feat(DynamicPricing) - Add dynamic charge model & validator

app/models/charge.rb

@@ -10,10 +10,10 @@ class Charge < ApplicationRecord
   belongs_to :billable_metric, -> { with_discarded }
 
   has_many :fees
-  has_many :filters, dependent: :destroy, class_name: 'ChargeFilter'
-  has_many :filter_values, through: :filters, class_name: 'ChargeFilterValue', source: :values
+  has_many :filters, dependent: :destroy, class_name: "ChargeFilter"
+  has_many :filter_values, through: :filters, class_name: "ChargeFilterValue", source: :values
 
-  has_many :applied_taxes, class_name: 'Charge::AppliedTax', dependent: :destroy
+  has_many :applied_taxes, class_name: "Charge::AppliedTax", dependent: :destroy
   has_many :taxes, through: :applied_taxes
 
   CHARGE_MODELS = %i[
@@ -24,6 +24,7 @@ class Charge < ApplicationRecord
     volume
     graduated_percentage
     custom
+    dynamic
   ].freeze
 
   REGROUPING_PAID_FEES_OPTIONS = %i[invoice].freeze
@@ -37,6 +38,7 @@ class Charge < ApplicationRecord
   validate :validate_percentage, if: -> { percentage? }
   validate :validate_volume, if: -> { volume? }
   validate :validate_graduated_percentage, if: -> { graduated_percentage? }
+  validate :validate_dynamic, if: -> { dynamic? }
 
   validates :min_amount_cents, numericality: {greater_than_or_equal_to: 0}, allow_nil: true
   validates :charge_model, presence: true
@@ -53,6 +55,10 @@ class Charge < ApplicationRecord
 
   scope :pay_in_advance, -> { where(pay_in_advance: true) }
 
+  def supports_grouped_by?
+    standard? || dynamic?
+  end
+
   private
 
   def validate_amount
@@ -79,6 +85,13 @@ def validate_graduated_percentage
     validate_charge_model(Charges::Validators::GraduatedPercentageService)
   end
 
+  def validate_dynamic
+    # Only sum aggregation is compatible with Dynamic Pricing for now
+    return if billable_metric.sum_agg?
+
+    errors.add(:charge_model, :invalid_aggregation_type_or_charge_model)
+  end
+
   def validate_charge_model(validator)
     instance = validator.new(charge: self)
     return if instance.valid?

app/models/clickhouse/events_raw.rb

@@ -10,11 +10,12 @@ class EventsRaw < BaseRecord
 #
 # Table name: events_raw
 #
-#  code                     :string           not null
-#  properties               :string           not null
-#  timestamp                :datetime         not null
-#  external_customer_id     :string           not null
-#  external_subscription_id :string           not null
-#  organization_id          :string           not null
-#  transaction_id           :string           not null
+#  code                       :string           not null
+#  precise_total_amount_cents :decimal(40, 15)
+#  properties                 :string           not null
+#  timestamp                  :datetime         not null
+#  external_customer_id       :string           not null
+#  external_subscription_id   :string           not null
+#  organization_id            :string           not null
+#  transaction_id             :string           not null
 #

app/models/events/common.rb

@@ -1,7 +1,7 @@
 # frozen_string_literal: true
 
 module Events
-  Common = Struct.new(:id, :organization_id, :transaction_id, :external_subscription_id, :timestamp, :code, :properties) do
+  Common = Struct.new(:id, :organization_id, :transaction_id, :external_subscription_id, :timestamp, :code, :properties, :precise_total_amount_cents) do
     def event_id
       id || transaction_id
     end

app/services/base_service.rb

@@ -170,6 +170,10 @@ def call_async(**args, &block)
     raise NotImplementedError
   end
 
+  protected
+
+  attr_writer :result
+
   private
 
   attr_reader :result, :source

app/services/billable_metrics/aggregations/base_service.rb

@@ -23,9 +23,16 @@ def initialize(event_store_class:, charge:, subscription:, boundaries:, filters:
       def aggregate(options: {})
         if grouped_by.present?
           compute_grouped_by_aggregation(options:)
+          if charge.dynamic?
+            compute_grouped_by_precise_total_amount_cents(options:)
+          end
         else
           compute_aggregation(options:)
+          if charge.dynamic?
+            compute_precise_total_amount_cents(options:)
+          end
         end
+        result
       end
 
       def compute_aggregation(options: {})
@@ -36,6 +43,14 @@ def compute_grouped_by_aggregation(options: {})
         raise NotImplementedError
       end
 
+      def compute_precise_total_amount_cents(options: {})
+        raise NotImplementedError
+      end
+
+      def compute_grouped_by_precise_total_amount_cents(options: {})
+        raise NotImplementedError
+      end
+
       def per_event_aggregation(exclude_event: false)
         Result.new.tap do |result|
           result.event_aggregation = compute_per_event_aggregation(exclude_event:)

app/services/billable_metrics/aggregations/sum_service.rb

@@ -4,7 +4,7 @@ module BillableMetrics
   module Aggregations
     class SumService < BillableMetrics::Aggregations::BaseService
       def initialize(...)
-        super(...)
+        super
 
         event_store.numeric_property = true
         event_store.aggregation_property = billable_metric.field_name
@@ -25,7 +25,7 @@ def compute_aggregation(options: {})
         result.options = {running_total: running_total(options)}
         result
       rescue ActiveRecord::StatementInvalid => e
-        result.service_failure!(code: 'aggregation_failure', message: e.message)
+        result.service_failure!(code: "aggregation_failure", message: e.message)
       end
 
       # NOTE: Apply the grouped_by filter to the aggregation
@@ -58,10 +58,25 @@ def compute_grouped_by_aggregation(options: {})
           group_result.count = count[:value] || 0
           group_result
         end
-
-        result
       rescue ActiveRecord::StatementInvalid => e
-        result.service_failure!(code: 'aggregation_failure', message: e.message)
+        result.service_failure!(code: "aggregation_failure", message: e.message)
+      end
+
+      def compute_precise_total_amount_cents(options: {})
+        result.precise_total_amount_cents = event_store.sum_precise_total_amount_cents
+        result.pay_in_advance_precise_total_amount_cents = event&.precise_total_amount_cents || 0
+      end
+
+      def compute_grouped_by_precise_total_amount_cents(options: {})
+        aggregations = event_store.grouped_sum_precise_total_amount_cents
+        return result if aggregations.blank?
+
+        aggregations.each do |aggregation|
+          group_result = result.aggregations.find { |group_result| group_result.grouped_by == aggregation[:groups] }
+          next unless group_result
+
+          group_result.precise_total_amount_cents = aggregation[:value]
+        end
       end
 
       # NOTE: Return cumulative sum of field_name based on the number of free units
@@ -91,8 +106,8 @@ def running_total_per_aggregation(aggregation)
       end
 
       def compute_pay_in_advance_aggregation
-        return BigDecimal(0) unless event
-        return BigDecimal(0) if event.properties.blank?
+        return BigDecimal("0") unless event
+        return BigDecimal("0") if event.properties.blank?
 
         value = event.properties.fetch(billable_metric.field_name, 0).to_s
 
@@ -103,7 +118,7 @@ def compute_pay_in_advance_aggregation
         )
 
         unless cached_aggregation
-          return_value = BigDecimal(value).negative? ? '0' : value
+          return_value = BigDecimal(value).negative? ? "0" : value
           handle_event_metadata(current_aggregation: value, max_aggregation: value, units_applied: value)
 
           return BigDecimal(return_value)

app/services/billable_metrics/prorated_aggregations/sum_service.rb

@@ -4,9 +4,9 @@ module BillableMetrics
   module ProratedAggregations
     class SumService < BillableMetrics::ProratedAggregations::BaseService
       def initialize(**args)
+        super
         @base_aggregator = BillableMetrics::Aggregations::SumService.new(**args)
-
-        super(**args)
+        @base_aggregator.result = result
 
         event_store.numeric_property = true
         event_store.aggregation_property = billable_metric.field_name

app/services/billable_metrics/prorated_aggregations/unique_count_service.rb

@@ -4,9 +4,10 @@ module BillableMetrics
   module ProratedAggregations
     class UniqueCountService < BillableMetrics::ProratedAggregations::BaseService
       def initialize(**args)
-        @base_aggregator = BillableMetrics::Aggregations::UniqueCountService.new(**args)
+        super
 
-        super(**args)
+        @base_aggregator = BillableMetrics::Aggregations::UniqueCountService.new(**args)
+        @base_aggregator.result = result
 
         event_store.aggregation_property = billable_metric.field_name
         event_store.use_from_boundary = !billable_metric.recurring

app/services/charges/apply_pay_in_advance_charge_model_service.rb

@@ -12,7 +12,7 @@ def initialize(charge:, aggregation_result:, properties:)
 
     def call
       unless charge.pay_in_advance?
-        return result.service_failure!(code: 'apply_charge_model_error', message: 'Charge is not pay_in_advance')
+        return result.service_failure!(code: "apply_charge_model_error", message: "Charge is not pay_in_advance")
       end
 
       amount = amount_including_event - amount_excluding_event
@@ -26,7 +26,7 @@ def call
       result.count = 1
       result.amount = amount_cents
       result.precise_amount = amount * currency.subunit_to_unit.to_d
-      result.unit_amount = rounded_amount.zero? ? BigDecimal(0) : rounded_amount / compute_units
+      result.unit_amount = rounded_amount.zero? ? BigDecimal("0") : rounded_amount / compute_units
       result
     end
 
@@ -48,15 +48,19 @@ def charge_model
         Charges::ChargeModels::PercentageService
       when :custom
         Charges::ChargeModels::CustomService
+      when :dynamic
+        Charges::ChargeModels::DynamicService
       else
         raise(NotImplementedError)
       end
     end
 
+    # Compute aggregation and apply charge for all events including the current one
     def amount_including_event
       @amount_including_event ||= charge_model.apply(charge:, aggregation_result:, properties:).amount
     end
 
+    # Compute aggregation and apply charge for all events excluding the current one
     def amount_excluding_event
       return @amount_excluding_event if defined?(@amount_excluding_event)
 
@@ -66,6 +70,12 @@ def amount_excluding_event
       previous_result.options = aggregation_result.options
       previous_result.aggregator = aggregation_result.aggregator
 
+      if aggregation_result.precise_total_amount_cents
+        previous_result.precise_total_amount_cents = (
+          aggregation_result.precise_total_amount_cents - aggregation_result.pay_in_advance_precise_total_amount_cents
+        )
+      end
+
       @amount_excluding_event ||= charge_model.apply(
         charge:,
         aggregation_result: previous_result,

app/services/charges/build_default_properties_service.rb

@@ -15,6 +15,7 @@ def call
       when :percentage then default_percentage_properties
       when :volume then default_volume_properties
       when :graduated_percentage then default_graduated_percentage_properties
+      when :dynamic then default_dynamic_properties
       end
     end
 
@@ -77,5 +78,9 @@ def default_graduated_percentage_properties
         ]
       }
     end
+
+    def default_dynamic_properties
+      {}
+    end
   end
 end

app/services/charges/charge_model_factory.rb

@@ -3,17 +3,18 @@
 module Charges
   class ChargeModelFactory
     def self.new_instance(charge:, aggregation_result:, properties:)
-      charge_model_class(charge:, aggregation_result:, properties:).new(charge:, aggregation_result:, properties:)
+      charge_model = charge_model_class(charge:)
+      if properties['grouped_by'].present? && !aggregation_result.aggregations.nil?
+        Charges::ChargeModels::GroupedService.new(charge_model: charge_model, charge:, aggregation_result:, properties:)
+      else
+        charge_model.new(charge:, aggregation_result:, properties:)
+      end
     end
 
-    def self.charge_model_class(charge:, aggregation_result:, properties:)
+    def self.charge_model_class(charge:)
       case charge.charge_model.to_sym
       when :standard
-        if properties['grouped_by'].present? && !aggregation_result.aggregations.nil?
-          Charges::ChargeModels::GroupedStandardService
-        else
-          Charges::ChargeModels::StandardService
-        end
+        Charges::ChargeModels::StandardService
       when :graduated
         if charge.prorated?
           Charges::ChargeModels::ProratedGraduatedService
@@ -30,6 +31,8 @@ def self.charge_model_class(charge:, aggregation_result:, properties:)
         Charges::ChargeModels::VolumeService
       when :custom
         Charges::ChargeModels::CustomService
+      when :dynamic
+        Charges::ChargeModels::DynamicService
       else
         raise(NotImplementedError)
       end

app/services/charges/charge_models/dynamic_service.rb

@@ -0,0 +1,31 @@
+# frozen_string_literal: true
+
+module Charges
+  module ChargeModels
+    class DynamicService < Charges::ChargeModels::BaseService
+      protected
+
+      def compute_amount
+        total_units = aggregation_result.full_units_number || units
+        return 0 if total_units.zero?
+
+        amount_cents = aggregation_result.precise_total_amount_cents
+        amount_cents / currency.subunit_to_unit
+      end
+
+      def unit_amount
+        # eventhough `full_units_number` is not set by the SumService, we still keep this code as is, to be future proof
+        total_units = aggregation_result.full_units_number || units
+        return 0 if total_units.zero?
+
+        compute_amount / total_units
+      end
+
+      private
+
+      def currency
+        charge.plan.amount.currency
+      end
+    end
+  end
+end

app/services/charges/charge_models/grouped_standard_service.rb → app/services/charges/charge_models/grouped_service.rb

@@ -2,22 +2,16 @@
 
 module Charges
   module ChargeModels
-    class GroupedStandardService < BaseService
-      def self.apply(...)
-        new(...).apply
-      end
-
-      def initialize(charge:, aggregation_result:, properties:)
-        super
+    class GroupedService < BaseService
+      def initialize(charge_model:, charge:, aggregation_result:, properties:)
+        super(charge:, aggregation_result:, properties:)
 
-        @charge = charge
-        @aggregation_result = aggregation_result
-        @properties = properties
+        @charge_model = charge_model
       end
 
       def apply
         result.grouped_results = aggregation_result.aggregations.map do |aggregation|
-          group_result = Charges::ChargeModels::StandardService.apply(
+          group_result = charge_model.apply(
             charge:,
             aggregation_result: aggregation,
             properties:
@@ -31,7 +25,7 @@ def apply
 
       protected
 
-      attr_accessor :charge, :aggregation_result, :properties
+      attr_accessor :charge_model
     end
   end
 end