feat(group-by): add grouped_by_value logic to event stores (#1617)

* feat(group-by): remove unused argument from event stores

* feat(group-by): add filters argument to event stores

* feat(group-by): add grouped_by_value logic to event stores

.gitignore

@@ -36,3 +36,4 @@ db/events_schema.rb
 LAGO_VERSION
 
 **/.DS_Store
+.vscode

app/services/billable_metrics/aggregations/base_service.rb

@@ -38,8 +38,7 @@ def event_store
           code: billable_metric.code,
           subscription:,
           boundaries:,
-          group:,
-          event:,
+          filters: { group: },
         )
       end
 

app/services/billable_metrics/aggregations/unique_count_service.rb

@@ -172,7 +172,7 @@ def base_scope
               from_datetime: subscription.started_at,
               to_datetime: subscription.terminated_at,
             },
-            group:,
+            filters: { group: },
           )
           store.aggregation_property = billable_metric.field_name
 

app/services/billable_metrics/aggregations/weighted_sum_service.rb

@@ -60,8 +60,7 @@ def latest_value_from_events
           code: billable_metric.code,
           subscription:,
           boundaries: { to_datetime: from_datetime },
-          group:,
-          event:,
+          filters: { group: },
         )
 
         event_store.use_from_boundary = false

app/services/billable_metrics/breakdown/sum_service.rb

@@ -27,8 +27,7 @@ def persisted_breakdown
           code: billable_metric.code,
           subscription:,
           boundaries: { to_datetime: from_datetime },
-          group:,
-          event:,
+          filters: { group: },
         )
 
         event_store.use_from_boundary = false

app/services/billable_metrics/prorated_aggregations/sum_service.rb

@@ -67,8 +67,7 @@ def persisted_sum
           code: billable_metric.code,
           subscription:,
           boundaries: { to_datetime: from_datetime },
-          group:,
-          event:,
+          filters: { group: },
         )
 
         event_store.use_from_boundary = false
@@ -93,8 +92,7 @@ def recurring_value
           code: billable_metric.code,
           subscription:,
           boundaries: { to_datetime: from_datetime },
-          group:,
-          event:,
+          filters: { group: },
         )
 
         event_store.use_from_boundary = false

app/services/events/stores/base_store.rb

@@ -3,18 +3,24 @@
 module Events
   module Stores
     class BaseStore
-      def initialize(code:, subscription:, boundaries:, group: nil, event: nil)
+      def initialize(code:, subscription:, boundaries:, filters: {})
         @code = code
         @subscription = subscription
         @boundaries = boundaries
-        @group = group
-        @event = event
+
+        @group = filters[:group]
+        @grouped_by = filters[:grouped_by]
+        @grouped_by_value = filters[:grouped_by_value]
 
         @aggregation_property = nil
         @numeric_property = false
         @use_from_boundary = true
       end
 
+      def grouped_by_value?
+        grouped_by.present? && grouped_by_value.present?
+      end
+
       def events(force_from: false)
         raise NotImplementedError
       end
@@ -72,7 +78,7 @@ def charges_duration
 
       protected
 
-      attr_accessor :code, :subscription, :group, :event, :boundaries
+      attr_accessor :code, :subscription, :group, :boundaries, :grouped_by, :grouped_by_value
 
       delegate :customer, to: :subscription
 

app/services/events/stores/clickhouse_store.rb

@@ -18,6 +18,8 @@ def events(force_from: false)
         scope = scope.where('events_raw.timestamp <= ?', to_datetime) if to_datetime
         scope = scope.where(numeric_condition) if numeric_property
 
+        scope = with_grouped_by_value(scope) if grouped_by_value?
+
         return scope unless group
 
         group_scope(scope)
@@ -176,6 +178,10 @@ def group_scope(scope)
         scope.where('events_raw.properties[?] = ?', group.parent.key.to_s => group.parent.value.to_s)
       end
 
+      def with_grouped_by_value(scope)
+        scope.where('events_raw.properties[?] = ?', grouped_by.to_s, grouped_by_value.to_s)
+      end
+
       def numeric_condition
         ActiveRecord::Base.sanitize_sql_for_conditions(
           [

app/services/events/stores/postgres_store.rb

@@ -16,6 +16,8 @@ def events(force_from: false)
             .where(numeric_condition)
         end
 
+        scope = with_grouped_by_value(scope) if grouped_by_value?
+
         return scope unless group
 
         group_scope(scope)
@@ -126,6 +128,10 @@ def group_scope(scope)
         scope.where('events.properties @> ?', { group.parent.key.to_s => group.parent.value }.to_json)
       end
 
+      def with_grouped_by_value(scope)
+        scope.where('events.properties @> ?', { grouped_by.to_s => grouped_by_value.to_s }.to_json)
+      end
+
       def sanitized_propery_name
         ActiveRecord::Base.sanitize_sql_for_conditions(
           ['events.properties->>?', aggregation_property],

spec/services/events/stores/clickhouse_store_spec.rb

@@ -8,8 +8,7 @@
       code:,
       subscription:,
       boundaries:,
-      group:,
-      event:,
+      filters: { group:, grouped_by:, grouped_by_value: },
     )
   end
 
@@ -31,14 +30,19 @@
   end
 
   let(:group) { nil }
-  let(:event) { nil }
+  let(:grouped_by) { nil }
+  let(:grouped_by_value) { nil }
 
   let(:events) do
     events = []
 
     5.times do |i|
       properties = { billable_metric.field_name => i + 1 }
-      properties[group.key.to_s] = group.value.to_s if group && i.even?
+
+      if i.even?
+        properties[group.key.to_s] = group.value.to_s if group
+        properties[grouped_by] = grouped_by_value if grouped_by_value
+      end
 
       events << Clickhouse::EventsRaw.create!(
         transaction_id: SecureRandom.uuid,
@@ -82,6 +86,15 @@
         expect(event_store.events.count).to eq(3)
       end
     end
+
+    context 'with grouped_by_value' do
+      let(:grouped_by) { 'region' }
+      let(:grouped_by_value) { 'europe' }
+
+      it 'returns a list of events' do
+        expect(event_store.events.count).to eq(3)
+      end
+    end
   end
 
   describe '.count' do

spec/services/events/stores/postgres_store_spec.rb

@@ -8,8 +8,7 @@
       code:,
       subscription:,
       boundaries:,
-      group:,
-      event:,
+      filters: { group:, grouped_by:, grouped_by_value: },
     )
   end
 
@@ -31,13 +30,14 @@
   end
 
   let(:group) { nil }
-  let(:event) { nil }
+  let(:grouped_by) { nil }
+  let(:grouped_by_value) { nil }
 
   let(:events) do
     events = []
 
     5.times do |i|
-      event = create(
+      event = build(
         :event,
         organization_id: organization.id,
         external_subscription_id: subscription.external_id,
@@ -49,11 +49,13 @@
         },
       )
 
-      if group && i.even?
-        event.properties[group.key] = group.value
-        event.save!
+      if i.even?
+        event.properties[group.key] = group.value if group
+        event.properties[grouped_by] = grouped_by_value if grouped_by_value
       end
 
+      event.save!
+
       events << event
     end
 
@@ -74,6 +76,15 @@
         expect(event_store.events.count).to eq(3)
       end
     end
+
+    context 'with grouped_by_value' do
+      let(:grouped_by) { 'region' }
+      let(:grouped_by_value) { 'europe' }
+
+      it 'returns a list of events' do
+        expect(event_store.events.count).to eq(3)
+      end
+    end
   end
 
   describe '.count' do