fix(api): normalize destination topics on write

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 {

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")))

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) {

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()