[ING-543] fix(events-processor): Select the same charge filter as Rails (#774)

Author: toommz

events-processor/models/flat_filters.go

@@ -177,51 +177,68 @@ func (ff *FlatFilter) ToDefaultFilter() *FlatFilter {
 	return defaultFilter
 }
 
+// MatchingFilter returns the filter the event belongs to, the charge's default bucket when none
+// matches.
+//
+// NOTE: This must return the same filter as the Ruby ChargeFilters::EventMatchingService, which
+// does the same job for organizations on the Postgres events store. Both the usage cache key and
+// the enriched event are built from it, so picking another one leaves the usage cache expiring a
+// key the usage reader never wrote to.
 func MatchingFilter(filters []FlatFilter, event *EnrichedEvent) *FlatFilter {
-	// Multiple filters are present, identify the best match
-	if len(filters) > 1 {
-		// First select all matching filters
-		matchingFilters := make([]FlatFilter, 0)
-		for _, filter := range filters {
-			if filter.HasFilters() && filter.IsMatchingEvent(event).Value() {
-				matchingFilters = append(matchingFilters, filter)
-			}
+	var bestFilter *FlatFilter
+
+	for i := range filters {
+		filter := &filters[i]
+		if !filter.HasFilters() || !filter.IsMatchingEvent(event).Value() {
+			continue
 		}
 
-		// No filters matches the event
-		if len(matchingFilters) == 0 {
-			// Return the charge's default bucket
-			return filters[0].ToDefaultFilter()
-
-		} else {
-			// NOTE: Multiple filters match the event (parent/child filters),
-			//       We must take only the one matching the most properties
-			var bestFilter *FlatFilter
-			for _, filter := range matchingFilters {
-				if bestFilter == nil {
-					bestFilter = &filter
-					continue
-				}
-
-				if len(filter.Filters.Keys()) > len(bestFilter.Filters.Keys()) {
-					bestFilter = &filter
-				}
-			}
-
-			// Return the best match
-			return bestFilter
+		if bestFilter == nil || filter.isBetterMatchThan(bestFilter) {
+			bestFilter = filter
 		}
+	}
 
-	} else {
-		filter := filters[0]
+	// No filter matches the event, it falls into the charge's default bucket
+	if bestFilter == nil {
+		return filters[0].ToDefaultFilter()
+	}
 
-		// Check if the only filter is matching the event
-		if filter.HasFilters() && filter.IsMatchingEvent(event).Value() {
-			// Return the only matching filter
-			return &filter
-		} else {
-			// Otherwise, return the charge's default bucket
-			return filter.ToDefaultFilter()
-		}
+	return bestFilter
+}
+
+// isBetterMatchThan reports whether ff must be preferred over other: it matches more of the event
+// properties, or as many but is older.
+//
+// NOTE: Ruby takes the first filter matching the most properties out of charge.filters, ordered by
+// the ChargeFilter default scope (order(updated_at: :asc)), hence the tie-break on
+// ChargeFilterUpdatedAt. The one on ChargeFilterID has no Ruby counterpart (Postgres resolves an
+// ORDER BY tie arbitrarily), it only keeps us deterministic when two filters share a timestamp,
+// rather than falling back to the order the filters were read in.
+func (ff *FlatFilter) isBetterMatchThan(other *FlatFilter) bool {
+	if keys, otherKeys := len(ff.Filters.Keys()), len(other.Filters.Keys()); keys != otherKeys {
+		return keys > otherKeys
+	}
+
+	updatedAt, otherUpdatedAt := ff.chargeFilterUpdatedAt(), other.chargeFilterUpdatedAt()
+	if !updatedAt.Equal(otherUpdatedAt) {
+		return updatedAt.Before(otherUpdatedAt)
 	}
+
+	return ff.chargeFilterID() < other.chargeFilterID()
+}
+
+func (ff *FlatFilter) chargeFilterUpdatedAt() time.Time {
+	if ff.ChargeFilterUpdatedAt == nil {
+		return time.Time{}
+	}
+
+	return *ff.ChargeFilterUpdatedAt
+}
+
+func (ff *FlatFilter) chargeFilterID() string {
+	if ff.ChargeFilterID == nil {
+		return ""
+	}
+
+	return *ff.ChargeFilterID
 }

events-processor/models/flat_filters_test.go

@@ -457,3 +457,66 @@ func TestMatchingFilter(t *testing.T) {
 		assert.Equal(t, result, flatFilter2)
 	})
 }
+
+func TestMatchingFilterTieBreak(t *testing.T) {
+	older := time.Now().Add(-time.Hour)
+	newer := time.Now()
+
+	buildFilter := func(id string, updatedAt time.Time, values FlatFilterValues) FlatFilter {
+		return FlatFilter{
+			OrganizationID:        "org_id",
+			BillableMetricCode:    "api_call",
+			PlanID:                "plan_id",
+			ChargeID:              "charge_id",
+			ChargeUpdatedAt:       updatedAt,
+			ChargeFilterID:        &id,
+			ChargeFilterUpdatedAt: &updatedAt,
+			Filters:               &values,
+		}
+	}
+
+	// NOTE: the flat_filters view expands a __ALL_FILTER_VALUES__ filter to the billable metric
+	//       filter values, so it becomes indistinguishable from an explicit filter when the metric
+	//       declares that single value. Ruby buckets the usage under the oldest of the two.
+	t.Run("it should return the oldest of two value-equivalent filters", func(t *testing.T) {
+		event := EnrichedEvent{
+			Properties: map[string]any{"model": "m1"},
+		}
+
+		allValues := buildFilter("all_values_filter_id", older, FlatFilterValues{"model": []string{"m1"}})
+		explicit := buildFilter("explicit_filter_id", newer, FlatFilterValues{"model": []string{"m1"}})
+
+		// The filters must not be picked out of the order they are read in
+		result := MatchingFilter([]FlatFilter{allValues, explicit}, &event)
+		assert.Equal(t, "all_values_filter_id", *result.ChargeFilterID)
+
+		result = MatchingFilter([]FlatFilter{explicit, allValues}, &event)
+		assert.Equal(t, "all_values_filter_id", *result.ChargeFilterID)
+	})
+
+	t.Run("it should return the smallest filter id when they share a timestamp", func(t *testing.T) {
+		event := EnrichedEvent{
+			Properties: map[string]any{"model": "m1"},
+		}
+
+		filter1 := buildFilter("filter_id1", older, FlatFilterValues{"model": []string{"m1"}})
+		filter2 := buildFilter("filter_id2", older, FlatFilterValues{"model": []string{"m1"}})
+
+		result := MatchingFilter([]FlatFilter{filter2, filter1}, &event)
+
+		assert.Equal(t, "filter_id1", *result.ChargeFilterID)
+	})
+
+	t.Run("it should return the most specific filter even when it is the newest", func(t *testing.T) {
+		event := EnrichedEvent{
+			Properties: map[string]any{"scheme": "visa", "method": "debit"},
+		}
+
+		parent := buildFilter("parent_filter_id", older, FlatFilterValues{"scheme": []string{"visa"}})
+		child := buildFilter("child_filter_id", newer, FlatFilterValues{"scheme": []string{"visa"}, "method": []string{"debit"}})
+
+		result := MatchingFilter([]FlatFilter{parent, child}, &event)
+
+		assert.Equal(t, "child_filter_id", *result.ChargeFilterID)
+	})
+}

events-processor/processors/events_processor/enrichment_service_test.go

@@ -1056,3 +1056,79 @@ func TestEvaluateExpression(t *testing.T) {
 		assert.Equal(t, "36", event.Properties["total_value"])
 	})
 }
+
+func TestEnrichEventWithOverlappingFilters(t *testing.T) {
+	// NOTE: the flat_filters view expands a __ALL_FILTER_VALUES__ filter to the billable metric
+	//       filter values, so it becomes indistinguishable from an explicit filter when the metric
+	//       declares that single value. The event must be enriched with the same filter the Ruby
+	//       usage reader buckets the usage under, the oldest one, otherwise the usage cache is
+	//       expired on a key that is never written to.
+	t.Run("With two value-equivalent filters on the same charge", func(t *testing.T) {
+		testEnv := setupEnrichmentTestEnv(t, false)
+		defer testEnv.Cleanup()
+
+		event := models.Event{
+			OrganizationID:         "1a901a90-1a90-1a90-1a90-1a901a901a90",
+			ExternalSubscriptionID: "sub_id",
+			Code:                   "api_calls",
+			Timestamp:              1741007009.0,
+			Properties:             map[string]any{"value": "12.12", "model": "m1"},
+			Source:                 "SQS",
+		}
+
+		bm := &models.BillableMetric{
+			ID:              "bm123",
+			OrganizationID:  event.OrganizationID,
+			Code:            event.Code,
+			AggregationType: models.AggregationTypeSum,
+			FieldName:       "value",
+			CreatedAt:       utils.NowNullTime(),
+			UpdatedAt:       utils.NowNullTime(),
+		}
+		testEnv.DataStore.SetBillableMetric(bm)
+
+		sub := &models.Subscription{
+			ID:             "sub123",
+			OrganizationID: &event.OrganizationID,
+			ExternalID:     event.ExternalSubscriptionID,
+			PlanID:         "plan_id",
+			StartedAt:      utils.NewNullTime(time.Unix(1700000000, 0)),
+		}
+		testEnv.DataStore.SetSubscription(sub)
+
+		older := time.Now().Add(-time.Hour)
+		newer := time.Now()
+
+		// The newest filter comes first, so reading the filters in order picks the wrong one
+		testEnv.DataStore.SetFlatFilters([]*models.FlatFilter{
+			{
+				OrganizationID:        event.OrganizationID,
+				BillableMetricCode:    event.Code,
+				PlanID:                "plan_id",
+				ChargeID:              "charge1",
+				ChargeUpdatedAt:       newer,
+				ChargeFilterID:        utils.StringPtr("explicit_filter_id"),
+				ChargeFilterUpdatedAt: &newer,
+				Filters:               &models.FlatFilterValues{"model": []string{"m1"}},
+			},
+			{
+				OrganizationID:        event.OrganizationID,
+				BillableMetricCode:    event.Code,
+				PlanID:                "plan_id",
+				ChargeID:              "charge1",
+				ChargeUpdatedAt:       newer,
+				ChargeFilterID:        utils.StringPtr("all_values_filter_id"),
+				ChargeFilterUpdatedAt: &older,
+				Filters:               &models.FlatFilterValues{"model": []string{"m1"}},
+			},
+		})
+
+		enrichResult := testEnv.EventProcessor.EnrichEvent(&event)
+		assert.True(t, enrichResult.Success())
+		assert.Equal(t, 1, len(enrichResult.Value()))
+
+		eventResult := enrichResult.Value()[0]
+		assert.Equal(t, "all_values_filter_id", *eventResult.ChargeFilterID)
+		assert.Equal(t, older.UTC(), eventResult.ChargeFilterUpdatedAt.UTC())
+	})
+}