diff --git a/internal/apirouter/destination_handlers.go b/internal/apirouter/destination_handlers.go index 450ab093..9926436d 100644 --- a/internal/apirouter/destination_handlers.go +++ b/internal/apirouter/destination_handlers.go @@ -109,6 +109,7 @@ func (h *DestinationHandlers) Create(c *gin.Context) { AbortWithValidationError(c, err) return } + destination.Topics = destination.Topics.Normalize() if err := h.registry.ValidateDestination(c.Request.Context(), &destination); err != nil { AbortWithValidationError(c, err) return @@ -183,6 +184,7 @@ func (h *DestinationHandlers) Update(c *gin.Context) { AbortWithValidationError(c, err) return } + updatedDestination.Topics = updatedDestination.Topics.Normalize() } shouldRevalidate := false if input.Type != "" && input.Type != originalDestination.Type { diff --git a/internal/apirouter/destination_handlers_test.go b/internal/apirouter/destination_handlers_test.go index 893e5400..00f50288 100644 --- a/internal/apirouter/destination_handlers_test.go +++ b/internal/apirouter/destination_handlers_test.go @@ -142,6 +142,46 @@ func TestAPI_Destinations(t *testing.T) { require.Equal(t, http.StatusUnprocessableEntity, resp.Code) }) + t.Run("duplicate topics are normalized", func(t *testing.T) { + h := newAPITest(t) + h.tenantStore.UpsertTenant(t.Context(), tf.Any(tf.WithID("t1"))) + + payload := validDestination() + payload["topics"] = []string{"user.created", "user.created"} + + req := h.jsonReq(http.MethodPost, "/api/v1/tenants/t1/destinations", payload) + resp := h.do(h.withAPIKey(req)) + + require.Equal(t, http.StatusCreated, resp.Code) + var dest destregistry.DestinationDisplay + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &dest)) + assert.Equal(t, models.Topics{"user.created"}, dest.Topics) + + stored, err := h.tenantStore.RetrieveDestination(t.Context(), "t1", dest.ID) + require.NoError(t, err) + assert.Equal(t, models.Topics{"user.created"}, stored.Topics) + }) + + t.Run("topic covered by sibling wildcard is normalized", func(t *testing.T) { + h := newAPITest(t, withTopicsAllowWildcards(true)) + h.tenantStore.UpsertTenant(t.Context(), tf.Any(tf.WithID("t1"))) + + payload := validDestination() + payload["topics"] = []string{"user.*", "user.created"} + + req := h.jsonReq(http.MethodPost, "/api/v1/tenants/t1/destinations", payload) + resp := h.do(h.withAPIKey(req)) + + require.Equal(t, http.StatusCreated, resp.Code) + var dest destregistry.DestinationDisplay + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &dest)) + assert.Equal(t, models.Topics{"user.*"}, dest.Topics) + + stored, err := h.tenantStore.RetrieveDestination(t.Context(), "t1", dest.ID) + require.NoError(t, err) + assert.Equal(t, models.Topics{"user.*"}, stored.Topics) + }) + t.Run("import timestamps", func(t *testing.T) { t.Run("disabled_at preserved on create", func(t *testing.T) { h := newAPITest(t) @@ -445,6 +485,50 @@ func TestAPI_Destinations(t *testing.T) { assert.Equal(t, models.Topics{"user.deleted"}, dest.Topics) }) + t.Run("duplicate topics are normalized", func(t *testing.T) { + h := newAPITest(t) + h.tenantStore.UpsertTenant(t.Context(), tf.Any(tf.WithID("t1"))) + h.tenantStore.CreateDestination(t.Context(), df.Any( + df.WithID("d1"), df.WithTenantID("t1"), df.WithTopics([]string{"user.created"}), + )) + + req := h.jsonReq(http.MethodPatch, "/api/v1/tenants/t1/destinations/d1", map[string]any{ + "topics": []string{"user.deleted", "user.deleted"}, + }) + resp := h.do(h.withAPIKey(req)) + + require.Equal(t, http.StatusOK, resp.Code) + var dest destregistry.DestinationDisplay + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &dest)) + assert.Equal(t, models.Topics{"user.deleted"}, dest.Topics) + + stored, err := h.tenantStore.RetrieveDestination(t.Context(), "t1", "d1") + require.NoError(t, err) + assert.Equal(t, models.Topics{"user.deleted"}, stored.Topics) + }) + + t.Run("topic covered by sibling wildcard is normalized", func(t *testing.T) { + h := newAPITest(t, withTopicsAllowWildcards(true)) + h.tenantStore.UpsertTenant(t.Context(), tf.Any(tf.WithID("t1"))) + h.tenantStore.CreateDestination(t.Context(), df.Any( + df.WithID("d1"), df.WithTenantID("t1"), df.WithTopics([]string{"user.created"}), + )) + + req := h.jsonReq(http.MethodPatch, "/api/v1/tenants/t1/destinations/d1", map[string]any{ + "topics": []string{"user.*", "user.created"}, + }) + resp := h.do(h.withAPIKey(req)) + + require.Equal(t, http.StatusOK, resp.Code) + var dest destregistry.DestinationDisplay + require.NoError(t, json.Unmarshal(resp.Body.Bytes(), &dest)) + assert.Equal(t, models.Topics{"user.*"}, dest.Topics) + + stored, err := h.tenantStore.RetrieveDestination(t.Context(), "t1", "d1") + require.NoError(t, err) + assert.Equal(t, models.Topics{"user.*"}, stored.Topics) + }) + t.Run("api key updates destination config", func(t *testing.T) { h := newAPITest(t) h.tenantStore.UpsertTenant(t.Context(), tf.Any(tf.WithID("t1"))) diff --git a/internal/models/entities.go b/internal/models/entities.go index 8d234308..100b66b5 100644 --- a/internal/models/entities.go +++ b/internal/models/entities.go @@ -197,6 +197,52 @@ func (t *Topics) Validate(availableTopics []string, allowWildcards bool) error { return nil } +// Normalize returns a topic set with redundant entries removed, preserving +// first-seen order. It performs two reductions: +// +// - exact duplicates are collapsed to their first occurrence +// (["user.created","user.created"] -> ["user.created"]). +// - an entry is folded away when a sibling wildcard pattern covers it and the +// entry does not itself cover that sibling +// (["user.*","user.created"] -> ["user.*"]). +// +// Two mutually-non-covering patterns are both kept +// (["*.created","user.*"] is unchanged), and ["*"] is returned as-is. +// Normalization never changes MatchTopic results: it only drops entries that +// are already covered by a retained entry. +func (t Topics) Normalize() Topics { + if t.MatchesAll() || len(t) <= 1 { + return t + } + result := make(Topics, 0, len(t)) + for _, e := range t { + if slices.Contains(result, e) { + continue // exact duplicate of an already-kept entry + } + if coveredByOther(e, t) { + continue // folded into a strictly-more-general sibling pattern + } + result = append(result, e) + } + return result +} + +// coveredByOther reports whether entry e is covered by some other entry p in +// topics such that p covers e but e does not cover p (p is strictly more +// general). This keeps mutually-covering entries and pattern pairs that neither +// covers, so only strictly-redundant entries are folded. +func coveredByOther(e string, topics Topics) bool { + for _, p := range topics { + if p == e { + continue + } + if matchTopicPattern(p, e) && !matchTopicPattern(e, p) { + return true + } + } + return false +} + func topicPatternMatchesAny(pattern string, topics []string) bool { for _, topic := range topics { if matchTopicPattern(pattern, topic) { diff --git a/internal/models/entities_test.go b/internal/models/entities_test.go index 86dce830..4f319b94 100644 --- a/internal/models/entities_test.go +++ b/internal/models/entities_test.go @@ -127,6 +127,61 @@ func TestDestinationTopics_Validate(t *testing.T) { } } +func TestDestinationTopics_Normalize(t *testing.T) { + t.Parallel() + + type testCase struct { + name string + topics models.Topics + expected models.Topics + } + + testCases := []testCase{ + { + name: "exact duplicates collapse to first occurrence", + topics: models.Topics{"user.created", "user.created"}, + expected: models.Topics{"user.created"}, + }, + { + name: "entry covered by sibling pattern is folded", + topics: models.Topics{"user.*", "user.created"}, + expected: models.Topics{"user.*"}, + }, + { + name: "covered entry folded, mutually-non-covering patterns kept", + topics: models.Topics{"*.created", "user.created", "user.*"}, + expected: models.Topics{"*.created", "user.*"}, + }, + { + name: "mutually-non-covering patterns are unchanged", + topics: models.Topics{"*.created", "user.*"}, + expected: models.Topics{"*.created", "user.*"}, + }, + { + name: "distinct non-redundant topics are unchanged", + topics: models.Topics{"user.created", "user.updated"}, + expected: models.Topics{"user.created", "user.updated"}, + }, + { + name: "match-all is returned as-is", + topics: models.Topics{"*"}, + expected: models.Topics{"*"}, + }, + { + name: "first-seen order is preserved", + topics: models.Topics{"user.updated", "user.created", "user.updated"}, + expected: models.Topics{"user.updated", "user.created"}, + }, + } + + for _, tc := range testCases { + t.Run(tc.name, func(t *testing.T) { + t.Parallel() + assert.Equal(t, tc.expected, tc.topics.Normalize()) + }) + } +} + func TestDestination_JSONMarshalWithDeliveryMetadata(t *testing.T) { t.Parallel()