feat(unique_count): Add active_unique_property to event stores

app/services/billable_metrics/prorated_aggregations/sum_service.rb

@@ -112,7 +112,7 @@ def per_event_aggregation
 
       protected
 
-      def persisted_event_store_instante
+      def persisted_event_store_instance
         event_store = event_store_class.new(
           code: billable_metric.code,
           subscription:,
@@ -137,7 +137,7 @@ def compute_event_aggregation
       end
 
       def persisted_sum
-        persisted_event_store_instante.prorated_sum(
+        persisted_event_store_instance.prorated_sum(
           period_duration:,
           persisted_duration: subscription.date_diff_with_timezone(from_datetime, to_datetime),
         )
@@ -147,7 +147,7 @@ def recurring_value
         previous_charge_fee_units = previous_charge_fee&.units
         return previous_charge_fee_units if previous_charge_fee_units
 
-        recurring_value_before_first_fee = persisted_event_store_instante.sum
+        recurring_value_before_first_fee = persisted_event_store_instance.sum
 
         ((recurring_value_before_first_fee || 0) <= 0) ? nil : recurring_value_before_first_fee
       end
@@ -174,7 +174,7 @@ def compute_grouped_event_aggregation
       end
 
       def grouped_persisted_sums
-        persisted_event_store_instante.grouped_prorated_sum(
+        persisted_event_store_instance.grouped_prorated_sum(
           period_duration:,
           persisted_duration: subscription.date_diff_with_timezone(from_datetime, to_datetime),
         )

app/services/events/stores/clickhouse_store.rb

@@ -108,6 +108,21 @@ def grouped_count
         prepare_grouped_result(::Clickhouse::EventsRaw.connection.select_all(sql).rows)
       end
 
+      # NOTE: check if an event created before the current on belongs to an active (as in present and not removed)
+      #       unique property
+      def active_unique_property?(event)
+        previous_event = events
+          .where('events_raw.properties[?] = ?', aggregation_property, event.properties[aggregation_property])
+          .where('events_raw.timestamp < ?', event.timestamp)
+          .reorder(timestamp: :desc)
+          .first
+
+        previous_event && (
+          previous_event.properties['operation_type'].nil? ||
+          previous_event.properties['operation_type'] == 'add'
+        )
+      end
+
       def unique_count
         query = Events::Stores::Clickhouse::UniqueCountQuery.new(store: self)
         sql = ActiveRecord::Base.sanitize_sql_for_conditions(

app/services/events/stores/postgres_store.rb

@@ -71,6 +71,21 @@ def grouped_count
         prepare_grouped_result(results)
       end
 
+      # NOTE: check if an event created before the current on belongs to an active (as in present and not removed)
+      #       unique property
+      def active_unique_property?(event)
+        previous_event = events.where.not(id: event.id)
+          .where('events.properties @> ?', { aggregation_property => event.properties[aggregation_property] }.to_json)
+          .where('events.timestamp < ?', event.timestamp)
+          .reorder(timestamp: :desc)
+          .first
+
+        previous_event && (
+          previous_event.properties['operation_type'].nil? ||
+          previous_event.properties['operation_type'] == 'add'
+        )
+      end
+
       def unique_count
         query = Events::Stores::Postgres::UniqueCountQuery.new(store: self)
         sql = ActiveRecord::Base.sanitize_sql_for_conditions([query.query])

app/services/utils/timezone_service.rb

@@ -10,7 +10,7 @@ def self.date_in_customer_timezone_sql(customer, value)
       end
       sanitized_timezone = ActiveRecord::Base.sanitize_sql_for_conditions(customer.applicable_timezone)
 
-      "#{sanitized_field_name}::timestamptz AT TIME ZONE '#{sanitized_timezone}'"
+      "(#{sanitized_field_name})::timestamptz AT TIME ZONE '#{sanitized_timezone}'"
     end
 
     def self.at_time_zone

spec/services/events/stores/clickhouse_store_spec.rb

@@ -151,6 +151,82 @@
     end
   end
 
+  describe '#active_unique_property?' do
+    before { event_store.aggregation_property = billable_metric.field_name }
+
+    it 'returns false when no previous events exist' do
+      event = ::Clickhouse::EventsRaw.create!(
+        organization_id: organization.id,
+        external_subscription_id: subscription.external_id,
+        external_customer_id: customer.external_id,
+        code:,
+        timestamp: (boundaries[:from_datetime] + 2.days).end_of_day,
+        properties: {
+          billable_metric.field_name => SecureRandom.uuid,
+        },
+      )
+
+      expect(event_store).not_to be_active_unique_property(event)
+    end
+
+    context 'when event is already active' do
+      it 'returns true if the event property is active' do
+        ::Clickhouse::EventsRaw.create!(
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 2.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+          },
+        )
+
+        event = ::Clickhouse::EventsRaw.create!(
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 3.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+          },
+        )
+
+        expect(event_store).to be_active_unique_property(event)
+      end
+    end
+
+    context 'with a previous removed event' do
+      it 'returns false' do
+        ::Clickhouse::EventsRaw.create!(
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 2.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+            operation_type: 'remove',
+          },
+        )
+
+        event = ::Clickhouse::EventsRaw.create!(
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 3.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+          },
+        )
+
+        expect(event_store).not_to be_active_unique_property(event)
+      end
+    end
+  end
+
   describe '#unique_count' do
     it 'returns the number of unique active event properties' do
       Clickhouse::EventsRaw.create!(

spec/services/events/stores/postgres_store_spec.rb

@@ -139,6 +139,87 @@
     end
   end
 
+  describe '#active_unique_property?' do
+    before { event_store.aggregation_property = billable_metric.field_name }
+
+    it 'returns false when no previous events exist' do
+      event = create(
+        :event,
+        organization_id: organization.id,
+        external_subscription_id: subscription.external_id,
+        external_customer_id: customer.external_id,
+        code:,
+        timestamp: (boundaries[:from_datetime] + 2.days).end_of_day,
+        properties: {
+          billable_metric.field_name => SecureRandom.uuid,
+        },
+      )
+
+      expect(event_store).not_to be_active_unique_property(event)
+    end
+
+    context 'when event is already active' do
+      it 'returns true if the event property is active' do
+        create(
+          :event,
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 2.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+          },
+        )
+
+        event = create(
+          :event,
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 3.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+          },
+        )
+
+        expect(event_store).to be_active_unique_property(event)
+      end
+    end
+
+    context 'with a previous removed event' do
+      it 'returns false' do
+        create(
+          :event,
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 2.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+            operation_type: 'remove',
+          },
+        )
+
+        event = create(
+          :event,
+          organization_id: organization.id,
+          external_subscription_id: subscription.external_id,
+          external_customer_id: customer.external_id,
+          code:,
+          timestamp: (boundaries[:from_datetime] + 3.days).end_of_day,
+          properties: {
+            billable_metric.field_name => 2,
+          },
+        )
+
+        expect(event_store).not_to be_active_unique_property(event)
+      end
+    end
+  end
+
   describe '#unique_count' do
     it 'returns the number of unique active event properties' do
       create(