refactor(opevents,apirouter): Event value + payload catalog for operator topics

Emit now takes an opevents.Event value (topic, tenant, data) instead of three
positional args, and payloads.go becomes the wire-contract catalog: the alert
payload shapes plus typed constructors, and apirouter's
tenant.subscription.updated payload moves there behind one too, replacing the
raw topic string + local struct. The alert monitor's consumer interface and
call sites are adapted mechanically.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>

Author: alexluong

internal/alert/monitor.go

@@ -15,7 +15,7 @@ import (
 
 // AlertEmitter is the interface for emitting alert events. Satisfied by opevents.Emitter.
 type AlertEmitter interface {
-	Emit(ctx context.Context, topic string, tenantID string, data any) error
+	Emit(ctx context.Context, ev opevents.Event) error
 }
 
 // DestinationDisabler handles disabling destinations.
@@ -198,7 +198,7 @@ func (m *alertMonitor) HandleAttempt(ctx context.Context, attempt DeliveryAttemp
 		}
 
 		emitFn := func(ctx context.Context) error {
-			if err := m.emitter.Emit(ctx, opevents.TopicAlertExhaustedRetries, attempt.Destination.TenantID, erData); err != nil {
+			if err := m.emitter.Emit(ctx, opevents.Event{Topic: opevents.TopicAlertExhaustedRetries, TenantID: attempt.Destination.TenantID, Data: erData}); err != nil {
 				return err
 			}
 			m.logger.Ctx(ctx).Audit("alert sent",
@@ -299,7 +299,7 @@ func (m *alertMonitor) handleConsecutiveFailure(ctx context.Context, attempt Del
 			Event:       attempt.Event,
 			Attempt:     attempt.Attempt,
 		}
-		if err := m.emitter.Emit(ctx, opevents.TopicAlertDestinationDisabled, attempt.Destination.TenantID, disabledData); err != nil {
+		if err := m.emitter.Emit(ctx, opevents.Event{Topic: opevents.TopicAlertDestinationDisabled, TenantID: attempt.Destination.TenantID, Data: disabledData}); err != nil {
 			return false, fmt.Errorf("failed to emit destination disabled alert: %w", err)
 		}
 	}
@@ -316,7 +316,7 @@ func (m *alertMonitor) handleConsecutiveFailure(ctx context.Context, attempt Del
 			Threshold: level,
 		},
 	}
-	if err := m.emitter.Emit(ctx, opevents.TopicAlertConsecutiveFailure, attempt.Destination.TenantID, cfData); err != nil {
+	if err := m.emitter.Emit(ctx, opevents.Event{Topic: opevents.TopicAlertConsecutiveFailure, TenantID: attempt.Destination.TenantID, Data: cfData}); err != nil {
 		return false, fmt.Errorf("failed to emit consecutive failure alert: %w", err)
 	}
 

internal/alert/monitor_test.go

@@ -12,6 +12,7 @@ import (
 
 	"github.com/hookdeck/outpost/internal/alert"
 	"github.com/hookdeck/outpost/internal/idempotence"
+	"github.com/hookdeck/outpost/internal/opevents"
 	"github.com/hookdeck/outpost/internal/models"
 	"github.com/hookdeck/outpost/internal/util/testutil"
 )
@@ -20,8 +21,8 @@ type mockAlertEmitter struct {
 	mock.Mock
 }
 
-func (m *mockAlertEmitter) Emit(ctx context.Context, topic string, tenantID string, data any) error {
-	args := m.Called(ctx, topic, tenantID, data)
+func (m *mockAlertEmitter) Emit(ctx context.Context, ev opevents.Event) error {
+	args := m.Called(ctx, ev.Topic, ev.TenantID, ev.Data)
 	return args.Error(0)
 }
 

internal/apirouter/destination_handlers.go

@@ -15,6 +15,7 @@ import (
 	"github.com/hookdeck/outpost/internal/idgen"
 	"github.com/hookdeck/outpost/internal/logging"
 	"github.com/hookdeck/outpost/internal/models"
+	"github.com/hookdeck/outpost/internal/opevents"
 	"github.com/hookdeck/outpost/internal/telemetry"
 	"github.com/hookdeck/outpost/internal/tenantstore"
 	"github.com/hookdeck/outpost/internal/util/maputil"
@@ -24,16 +25,7 @@ import (
 // SubscriptionEmitter emits operator events for subscription changes.
 // Satisfied by opevents.Emitter.
 type SubscriptionEmitter interface {
-	Emit(ctx context.Context, topic string, tenantID string, data any) error
-}
-
-// TenantSubscriptionUpdatedData is the data payload for tenant.subscription.updated events.
-type TenantSubscriptionUpdatedData struct {
-	TenantID                  string   `json:"tenant_id"`
-	Topics                    []string `json:"topics"`
-	PreviousTopics            []string `json:"previous_topics"`
-	DestinationsCount         int      `json:"destinations_count"`
-	PreviousDestinationsCount int      `json:"previous_destinations_count"`
+	Emit(ctx context.Context, ev opevents.Event) error
 }
 
 type DestinationHandlers struct {
@@ -572,14 +564,13 @@ func (h *DestinationHandlers) emitSubscriptionUpdateIfChanged(ctx context.Contex
 		return
 	}
 
-	data := TenantSubscriptionUpdatedData{
+	if err := h.emitter.Emit(ctx, opevents.TenantSubscriptionUpdatedEvent(opevents.TenantSubscriptionUpdatedData{
 		TenantID:                  tenantID,
 		Topics:                    tenant.Topics,
 		PreviousTopics:            prev.topics,
 		DestinationsCount:         tenant.DestinationsCount,
 		PreviousDestinationsCount: prev.destinationsCount,
-	}
-	if err := h.emitter.Emit(ctx, "tenant.subscription.updated", tenantID, data); err != nil {
+	})); err != nil {
 		h.logger.Ctx(ctx).Error("failed to emit subscription update", zap.Error(err))
 	}
 }

internal/apirouter/destination_handlers_test.go

@@ -7,9 +7,9 @@ import (
 	"testing"
 	"time"
 
-	"github.com/hookdeck/outpost/internal/apirouter"
 	"github.com/hookdeck/outpost/internal/destregistry"
 	"github.com/hookdeck/outpost/internal/models"
+	"github.com/hookdeck/outpost/internal/opevents"
 	"github.com/hookdeck/outpost/internal/tenantstore"
 	"github.com/stretchr/testify/assert"
 	"github.com/stretchr/testify/require"
@@ -1279,7 +1279,7 @@ func TestAPI_SubscriptionUpdated(t *testing.T) {
 		assert.Equal(t, "tenant.subscription.updated", call.topic)
 		assert.Equal(t, "t1", call.tenantID)
 
-		data := call.data.(apirouter.TenantSubscriptionUpdatedData)
+		data := call.data.(opevents.TenantSubscriptionUpdatedData)
 		assert.Equal(t, 1, data.DestinationsCount)
 		assert.Equal(t, 0, data.PreviousDestinationsCount)
 	})
@@ -1295,7 +1295,7 @@ func TestAPI_SubscriptionUpdated(t *testing.T) {
 		require.Equal(t, http.StatusOK, resp.Code)
 		require.Len(t, h.subscriptionEmitter.calls, 1)
 
-		data := h.subscriptionEmitter.calls[0].data.(apirouter.TenantSubscriptionUpdatedData)
+		data := h.subscriptionEmitter.calls[0].data.(opevents.TenantSubscriptionUpdatedData)
 		assert.Equal(t, 0, data.DestinationsCount)
 		assert.Equal(t, 1, data.PreviousDestinationsCount)
 	})
@@ -1315,7 +1315,7 @@ func TestAPI_SubscriptionUpdated(t *testing.T) {
 		require.Equal(t, http.StatusOK, resp.Code)
 		require.Len(t, h.subscriptionEmitter.calls, 1)
 
-		data := h.subscriptionEmitter.calls[0].data.(apirouter.TenantSubscriptionUpdatedData)
+		data := h.subscriptionEmitter.calls[0].data.(opevents.TenantSubscriptionUpdatedData)
 		assert.Contains(t, data.Topics, "user.deleted")
 		assert.Contains(t, data.PreviousTopics, "user.created")
 	})

internal/apirouter/router_test.go

@@ -16,6 +16,7 @@ import (
 	"github.com/hookdeck/outpost/internal/logging"
 	"github.com/hookdeck/outpost/internal/logstore"
 	"github.com/hookdeck/outpost/internal/models"
+	"github.com/hookdeck/outpost/internal/opevents"
 	"github.com/hookdeck/outpost/internal/portal"
 	"github.com/hookdeck/outpost/internal/publishmq"
 	"github.com/hookdeck/outpost/internal/telemetry"
@@ -244,8 +245,8 @@ type emitCall struct {
 	data     any
 }
 
-func (m *mockSubscriptionEmitter) Emit(_ context.Context, topic string, tenantID string, data any) error {
-	m.calls = append(m.calls, emitCall{topic: topic, tenantID: tenantID, data: data})
+func (m *mockSubscriptionEmitter) Emit(_ context.Context, ev opevents.Event) error {
+	m.calls = append(m.calls, emitCall{topic: ev.Topic, tenantID: ev.TenantID, data: ev.Data})
 	return nil
 }
 

internal/opevents/emitter.go

@@ -15,9 +15,17 @@ const (
 	backoffFactor = 2
 )
 
+// Event is a request to emit an operator event. Callers (alert eval, apirouter)
+// build it and hand it to Emit, which owns envelope construction and delivery.
+type Event struct {
+	Topic    string
+	TenantID string
+	Data     any
+}
+
 // Emitter is the interface for emitting operator events.
 type Emitter interface {
-	Emit(ctx context.Context, topic string, tenantID string, data any) error
+	Emit(ctx context.Context, ev Event) error
 }
 
 // emitter is the default Emitter implementation.
@@ -54,23 +62,23 @@ func NewEmitter(sink Sink, deploymentID string, topics []string) Emitter {
 	}
 }
 
-func (e *emitter) Emit(ctx context.Context, topic string, tenantID string, data any) error {
+func (e *emitter) Emit(ctx context.Context, ev Event) error {
 	// Topic filtering: nil filter means accept all ("*")
-	if e.topicFilter != nil && !e.topicFilter[topic] {
+	if e.topicFilter != nil && !e.topicFilter[ev.Topic] {
 		return nil
 	}
 
-	rawData, err := json.Marshal(data)
+	rawData, err := json.Marshal(ev.Data)
 	if err != nil {
 		return fmt.Errorf("opevents: failed to marshal data: %w", err)
 	}
 
 	event := &OperatorEvent{
 		ID:           idgen.String(),
-		Topic:        topic,
+		Topic:        ev.Topic,
 		Time:         time.Now(),
 		DeploymentID: e.deploymentID,
-		TenantID:     tenantID,
+		TenantID:     ev.TenantID,
 		Data:         rawData,
 	}
 
@@ -104,6 +112,6 @@ func (e *emitter) sendWithRetry(ctx context.Context, event *OperatorEvent) error
 // noopEmitter discards all events. Used when operator events are disabled.
 type noopEmitter struct{}
 
-func (e *noopEmitter) Emit(ctx context.Context, topic string, tenantID string, data any) error {
+func (e *noopEmitter) Emit(ctx context.Context, ev Event) error {
 	return nil
 }

internal/opevents/emitter_test.go

@@ -51,7 +51,7 @@ func TestEmitter_Emit(t *testing.T) {
 		sink := &mockSink{}
 		em := opevents.NewEmitter(sink, "deploy-1", []string{"*"})
 
-		err := em.Emit(context.Background(), "any.topic", "tenant-1", map[string]string{"key": "val"})
+		err := em.Emit(context.Background(), opevents.Event{Topic: "any.topic", TenantID: "tenant-1", Data: map[string]string{"key": "val"}})
 		require.NoError(t, err)
 
 		events := sink.sentEvents()
@@ -69,7 +69,7 @@ func TestEmitter_Emit(t *testing.T) {
 			opevents.TopicTenantSubscriptionUpdated,
 		})
 
-		err := em.Emit(context.Background(), opevents.TopicAlertConsecutiveFailure, "t1", "data")
+		err := em.Emit(context.Background(), opevents.Event{Topic: opevents.TopicAlertConsecutiveFailure, TenantID: "t1", Data: "data"})
 		require.NoError(t, err)
 		assert.Len(t, sink.sentEvents(), 1)
 	})
@@ -79,7 +79,7 @@ func TestEmitter_Emit(t *testing.T) {
 		sink := &mockSink{}
 		em := opevents.NewEmitter(sink, "", []string{opevents.TopicAlertConsecutiveFailure})
 
-		err := em.Emit(context.Background(), opevents.TopicTenantSubscriptionUpdated, "t1", "data")
+		err := em.Emit(context.Background(), opevents.Event{Topic: opevents.TopicTenantSubscriptionUpdated, TenantID: "t1", Data: "data"})
 		require.NoError(t, err)
 		assert.Empty(t, sink.sentEvents())
 	})
@@ -93,7 +93,7 @@ func TestEmitter_Emit(t *testing.T) {
 			Count int `json:"count"`
 		}
 
-		err := em.Emit(context.Background(), "test.topic", "tenant-42", payload{Count: 7})
+		err := em.Emit(context.Background(), opevents.Event{Topic: "test.topic", TenantID: "tenant-42", Data: payload{Count: 7}})
 		require.NoError(t, err)
 
 		events := sink.sentEvents()
@@ -116,7 +116,7 @@ func TestEmitter_Emit(t *testing.T) {
 		sink := &mockSink{}
 		em := opevents.NewEmitter(sink, "", []string{"*"})
 
-		err := em.Emit(context.Background(), "t", "tenant", "data")
+		err := em.Emit(context.Background(), opevents.Event{Topic: "t", TenantID: "tenant", Data: "data"})
 		require.NoError(t, err)
 
 		events := sink.sentEvents()
@@ -134,7 +134,7 @@ func TestEmitter_Emit(t *testing.T) {
 		sink := &mockSink{}
 		em := opevents.NewEmitter(sink, "", []string{})
 
-		err := em.Emit(context.Background(), "any.topic", "t1", "data")
+		err := em.Emit(context.Background(), opevents.Event{Topic: "any.topic", TenantID: "t1", Data: "data"})
 		require.NoError(t, err)
 		assert.Empty(t, sink.sentEvents(), "noop emitter should not send events")
 	})
@@ -150,7 +150,7 @@ func TestEmitter_Emit(t *testing.T) {
 		}
 		em := opevents.NewEmitter(sink, "", []string{"*"})
 
-		err := em.Emit(context.Background(), "t", "t1", "data")
+		err := em.Emit(context.Background(), opevents.Event{Topic: "t", TenantID: "t1", Data: "data"})
 		require.NoError(t, err)
 		assert.Len(t, sink.sentEvents(), 1)
 	})
@@ -166,7 +166,7 @@ func TestEmitter_Emit(t *testing.T) {
 		}
 		em := opevents.NewEmitter(sink, "", []string{"*"})
 
-		err := em.Emit(context.Background(), "t", "t1", "data")
+		err := em.Emit(context.Background(), opevents.Event{Topic: "t", TenantID: "t1", Data: "data"})
 		require.Error(t, err)
 		assert.Contains(t, err.Error(), "fail-3")
 		assert.Empty(t, sink.sentEvents())
@@ -182,7 +182,7 @@ func TestEmitter_Emit(t *testing.T) {
 		}
 		em := opevents.NewEmitter(sink, "", []string{"*"})
 
-		err := em.Emit(ctx, "t", "t1", "data")
+		err := em.Emit(ctx, opevents.Event{Topic: "t", TenantID: "t1", Data: "data"})
 		require.Error(t, err)
 	})
 }