feat(flexible aggregation) - Support flexible aggregation on billable metric

.github/workflows/spec.yml

@@ -2,7 +2,7 @@ name: Run Spec
 on:
   push:
     branches:
-      - "main"
+      - 'main'
   pull_request:
     types: [opened, synchronize, reopened]
 jobs:
@@ -13,7 +13,7 @@ jobs:
       postgres:
         image: postgres:14-alpine
         ports:
-          - "5432:5432"
+          - '5432:5432'
         env:
           POSTGRES_DB: lago
           POSTGRES_USER: lago
@@ -30,8 +30,8 @@ jobs:
 
     env:
       RAILS_ENV: test
-      DATABASE_URL: "postgres://lago:lago@localhost:5432/lago"
-      LAGO_REDIS_CACHE_URL: "redis://localhost:6379"
+      DATABASE_URL: 'postgres://lago:lago@localhost:5432/lago'
+      LAGO_REDIS_CACHE_URL: 'redis://localhost:6379'
       RAILS_MASTER_KEY: ${{ secrets.RAILS_TEST_KEY }}
       SECRET_KEY_BASE: ${{ secrets.SECRET_KEY_BASE }}
       LAGO_API_URL: https://api.lago.dev
@@ -42,16 +42,16 @@ jobs:
       LAGO_CLICKHOUSE_MIGRATIONS_ENABLED: true
       LAGO_CLICKHOUSE_HOST: localhost
       LAGO_CLICKHOUSE_DATABASE: default
-      LAGO_CLICKHOUSE_USERNAME: ""
-      LAGO_CLICKHOUSE_PASSWORD: ""
+      LAGO_CLICKHOUSE_USERNAME: ''
+      LAGO_CLICKHOUSE_PASSWORD: ''
 
     steps:
       - name: Checkout code
         uses: actions/checkout@v4
       - name: Install Ruby and gems
         uses: ruby/setup-ruby@v1
         with:
-          ruby-version: "3.3.4"
+          ruby-version: '3.3.4'
           bundler-cache: true
       - name: Start Clickhouse database
         run: |

Dockerfile

@@ -2,12 +2,14 @@ FROM ruby:3.3.4-slim AS build
 
 WORKDIR /app
 
+RUN apt update -qq && apt install nodejs curl build-essential git pkg-config libpq-dev libclang-dev curl -y && \
+  curl https://sh.rustup.rs -sSf | bash -s -- -y
+
 COPY ./Gemfile /app/Gemfile
 COPY ./Gemfile.lock /app/Gemfile.lock
 
-RUN apt update -qq && apt install nodejs build-essential git pkg-config libpq-dev curl -y
-
 ENV BUNDLER_VERSION='2.5.5'
+ENV PATH="$PATH:/root/.cargo/bin/"
 RUN gem install bundler --no-document -v '2.5.5'
 
 RUN bundle config build.nokogiri --use-system-libraries &&\

Dockerfile.dev

@@ -2,15 +2,19 @@ FROM ruby:3.3.4-slim
 
 WORKDIR /app
 
+RUN apt update -qq && \
+  apt install nodejs build-essential curl git pkg-config libpq-dev libclang-dev -y && \
+  curl https://sh.rustup.rs -sSf | bash -s -- -y
+
+ENV BUNDLER_VERSION='2.5.5'
+ENV PATH="$PATH:/root/.cargo/bin/"
+
 COPY ./Gemfile /app/Gemfile
 COPY ./Gemfile.lock /app/Gemfile.lock
 
-RUN apt update -qq && apt install nodejs build-essential git pkg-config libpq-dev -y
-
-ENV BUNDLER_VERSION='2.5.5'
 RUN gem install bundler --no-document -v '2.5.5'
 
-RUN bundle config build.nokogiri --use-system-libraries &&\
- bundle install
+RUN bundle config build.nokogiri --use-system-libraries && \
+  bundle install
 
 CMD ["./scripts/start.dev.sh"]

Gemfile

@@ -89,6 +89,8 @@ gem "valvat", require: false
 # Data Export
 gem "csv", "~> 3.0"
 
+gem 'lago-expression', github: 'getlago/lago-expression', glob: 'expression-ruby/lago-expression.gemspec'
+
 group :development, :test, :staging do
   gem "factory_bot_rails"
   gem "faker"

Gemfile.lock

@@ -1,3 +1,14 @@
+GIT
+  remote: https://github.com/getlago/lago-expression.git
+  revision: a096edf11524bfda7d7f196ddb7722b6c5b8eeae
+  glob: expression-ruby/lago-expression.gemspec
+  specs:
+    lago-expression (0.0.1)
+      bigdecimal
+      rake (~> 13)
+      rake-compiler (~> 1.2)
+      rb_sys (~> 0.9.63)
+
 GIT
   remote: https://github.com/glebm/i18n-tasks.git
   revision: 2cba1093e3c555b6664f62604a2e2f2dfe6f1a6e
@@ -640,13 +651,16 @@ GEM
       zeitwerk (~> 2.6)
     rainbow (3.1.1)
     rake (13.2.1)
+    rake-compiler (1.2.8)
+      rake
     ransack (4.1.1)
       activerecord (>= 6.1.5)
       activesupport (>= 6.1.5)
       i18n
     rb-fsevent (0.11.2)
     rb-inotify (0.11.1)
       ffi (~> 1.0)
+    rb_sys (0.9.102)
     rbs (3.6.1)
       logger
     rdoc (6.7.0)
@@ -898,6 +912,7 @@ DEPENDENCIES
   kaminari-activerecord
   karafka (~> 2.4.0)
   karafka-web (~> 0.9.0)
+  lago-expression!
   lograge
   logstash-event
   money-rails

app/controllers/api/v1/billable_metrics_controller.rb

@@ -92,6 +92,24 @@ def index
         )
       end
 
+      def evaluate_expression
+        result = ::BillableMetrics::EvaluateExpressionService.call(
+          expression: params[:expression],
+          event: expression_event_params[:event]
+        )
+
+        if result.success?
+          render(
+            json: ::V1::BillableMetricExpressionResultSerializer.new(
+              result,
+              root_name: "expression_result"
+            )
+          )
+        else
+          render_error_response(result)
+        end
+      end
+
       private
 
       def input_params
@@ -103,9 +121,18 @@ def input_params
           :weighted_interval,
           :recurring,
           :field_name,
+          :expression,
           filters: [:key, {values: []}]
         )
       end
+
+      def expression_event_params
+        params.permit(event: [
+          :code,
+          :timestamp,
+          properties: {}
+        ])
+      end
     end
   end
 end

app/graphql/types/billable_metrics/create_input.rb

@@ -8,6 +8,7 @@ class CreateInput < BaseInputObject
       argument :aggregation_type, Types::BillableMetrics::AggregationTypeEnum, required: true
       argument :code, String, required: true
       argument :description, String
+      argument :expression, String, required: false
       argument :field_name, String, required: false
       argument :name, String, required: true
       argument :recurring, Boolean, required: false

app/graphql/types/billable_metrics/object.rb

@@ -15,6 +15,7 @@ class Object < Types::BaseObject
       field :description, String
 
       field :aggregation_type, Types::BillableMetrics::AggregationTypeEnum, null: false
+      field :expression, String, null: true
       field :field_name, String, null: true
       field :weighted_interval, Types::BillableMetrics::WeightedIntervalEnum, null: true
 

app/graphql/types/billable_metrics/update_input.rb

@@ -10,6 +10,7 @@ class UpdateInput < BaseInputObject
       argument :aggregation_type, Types::BillableMetrics::AggregationTypeEnum, required: true
       argument :code, String, required: true
       argument :description, String
+      argument :expression, String, required: false
       argument :field_name, String, required: false
       argument :name, String, required: true
       argument :recurring, Boolean, required: false

app/models/billable_metric.rb

@@ -33,6 +33,7 @@ class BillableMetric < ApplicationRecord
   enum weighted_interval: WEIGHTED_INTERVAL
 
   validate :validate_recurring
+  validate :validate_expression
 
   validates :name, presence: true
   validates :field_name, presence: true, if: :should_have_field_name?
@@ -47,6 +48,8 @@ class BillableMetric < ApplicationRecord
 
   default_scope -> { kept }
 
+  scope :with_expression, -> { where("expression IS NOT NULL AND expression <> ''") }
+
   def self.ransackable_attributes(_auth_object = nil)
     %w[name code]
   end
@@ -75,6 +78,13 @@ def validate_recurring
 
     errors.add(:recurring, :not_compatible_with_aggregation_type)
   end
+
+  def validate_expression
+    return if expression.blank?
+    return if Lago::ExpressionParser.validate(expression).blank?
+
+    errors.add(:expression, :invalid_expression)
+  end
 end
 
 # == Schema Information
@@ -87,6 +97,7 @@ def validate_recurring
 #  custom_aggregator :text
 #  deleted_at        :datetime
 #  description       :string
+#  expression        :string
 #  field_name        :string
 #  name              :string           not null
 #  properties        :jsonb
@@ -99,6 +110,7 @@ def validate_recurring
 # Indexes
 #
 #  index_billable_metrics_on_deleted_at                (deleted_at)
+#  index_billable_metrics_on_org_id_and_code_and_expr  (organization_id,code,expression) WHERE ((expression IS NOT NULL) AND ((expression)::text <> ''::text))
 #  index_billable_metrics_on_organization_id           (organization_id)
 #  index_billable_metrics_on_organization_id_and_code  (organization_id,code) UNIQUE WHERE (deleted_at IS NULL)
 #

app/serializers/v1/billable_metric_expression_result_serializer.rb

@@ -0,0 +1,9 @@
+# frozen_string_literal: true
+
+module V1
+  class BillableMetricExpressionResultSerializer < ModelSerializer
+    def serialize
+      {value: model.evaluation_result}
+    end
+  end
+end

app/serializers/v1/billable_metric_serializer.rb

@@ -13,6 +13,7 @@ def serialize
         recurring: model.recurring,
         created_at: model.created_at.iso8601,
         field_name: model.field_name,
+        expression: model.expression,
         active_subscriptions_count:,
         draft_invoices_count:,
         plans_count:

app/services/billable_metrics/create_service.rb

@@ -23,7 +23,8 @@ def call
           recurring: args[:recurring] || false,
           aggregation_type: args[:aggregation_type]&.to_sym,
           field_name: args[:field_name],
-          weighted_interval: args[:weighted_interval]&.to_sym
+          weighted_interval: args[:weighted_interval]&.to_sym,
+          expression: args[:expression]
         )
 
         if args[:filters].present?

app/services/billable_metrics/destroy_service.rb

@@ -10,6 +10,8 @@ def initialize(metric:)
     def call
       return result.not_found_failure!(resource: 'billable_metric') unless metric
 
+      BillableMetrics::ExpressionCacheService.expire_cache(metric.organization.id, metric.code)
+
       draft_invoice_ids = Invoice.draft.joins(plans: [:billable_metrics])
         .where(billable_metrics: {id: metric.id}).distinct.pluck(:id)
 

app/services/billable_metrics/evaluate_expression_service.rb

@@ -0,0 +1,37 @@
+# frozen_string_literal: true
+
+module BillableMetrics
+  class EvaluateExpressionService < BaseService
+    def initialize(expression:, event:)
+      @expression = expression
+      @event = event || {}
+      super
+    end
+
+    def call
+      if expression.blank?
+        return result.single_validation_failure!(field: 'expression', error_code: 'value_is_mandatory')
+      end
+
+      expression_validation_result = Lago::ExpressionParser.validate(expression)
+      if expression_validation_result.present?
+        return result.single_validation_failure!(field: 'expression', error_code: 'invalid_expression')
+      end
+
+      evaluation_event = Lago::Event.new(
+        event['code'].to_s,
+        event['timestamp'].to_i || Time.current.to_i,
+        event['properties']&.transform_values(&:to_s) || {}
+      )
+
+      result.evaluation_result = Lago::ExpressionParser.parse(expression).evaluate(evaluation_event)
+      result
+    rescue RuntimeError
+      result.single_validation_failure!(field: 'event', error_code: 'invalid_event')
+    end
+
+    private
+
+    attr_reader :expression, :event
+  end
+end

app/services/billable_metrics/expression_cache_service.rb

@@ -0,0 +1,27 @@
+# frozen_string_literal: true
+
+module BillableMetrics
+  class ExpressionCacheService < CacheService
+    CACHE_KEY_VERSION = "1"
+
+    def initialize(organization_id, billable_metric_code)
+      @organization_id = organization_id
+      @billable_metric_code = billable_metric_code
+
+      super
+    end
+
+    def cache_key
+      [
+        'expression',
+        CACHE_KEY_VERSION,
+        organization_id,
+        billable_metric_code
+      ].compact.join('/')
+    end
+
+    private
+
+    attr_reader :organization_id, :billable_metric_code
+  end
+end

app/services/billable_metrics/update_service.rb

@@ -39,6 +39,11 @@ def call
         billable_metric.field_name = params[:field_name] if params.key?(:field_name)
         billable_metric.recurring = params[:recurring] if params.key?(:recurring)
         billable_metric.weighted_interval = params[:weighted_interval]&.to_sym if params.key?(:weighted_interval)
+        billable_metric.expression = params[:expression] if params.key?(:expression)
+
+        if params.key?(:expression) || params.key?(:field_name)
+          BillableMetrics::ExpressionCacheService.expire_cache(organization.id, billable_metric.code)
+        end
       end
 
       billable_metric.save!

app/services/cache_service.rb

@@ -0,0 +1,28 @@
+# frozen_string_literal: true
+
+class CacheService < BaseService
+  def initialize(*, expires_in: nil)
+    @expires_in = expires_in
+    super(nil)
+  end
+
+  def self.expire_cache(*, **)
+    new(*, **).expire_cache
+  end
+
+  def cache_key
+    raise NotImplementedError
+  end
+
+  def call(&)
+    Rails.cache.fetch(cache_key, expires_in:, &)
+  end
+
+  def expire_cache
+    Rails.cache.delete(cache_key)
+  end
+
+  private
+
+  attr_reader :expires_in
+end