fix: warn instead of crash when operator events topics configured without sink (#835)

* fix: warn instead of crash when operator events topics configured without sink

When topics are configured but no sink is set, the server would crash on
startup with a fatal error. Instead, log a warning and return a NoopSink
so the server can start gracefully (operator events will be dropped).

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

* chore: gofmt

* refactor: use project logger instead of slog in opevents.NewSink

Replace stdlib log/slog with the project's *logging.Logger for
consistency. NewSink now accepts a logger parameter, and callers in
builder.go pass b.logger.

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

---------

Co-authored-by: Claude <noreply@anthropic.com>

Author: alexluong

cmd/outpost/migrate.go

@@ -70,8 +70,8 @@ func newMigrateCommand() *cli.Command {
 				Action: runMigrateVerify,
 			},
 			{
-				Name:   "unlock",
-				Usage:  "Force clear the Redis migration lock (use with caution)",
+				Name:  "unlock",
+				Usage: "Force clear the Redis migration lock (use with caution)",
 				Flags: []cli.Flag{
 					&cli.BoolFlag{
 						Name:    "yes",

internal/config/config.go

@@ -59,7 +59,7 @@ type Config struct {
 	DeploymentID        string   `yaml:"deployment_id" env:"DEPLOYMENT_ID" desc:"Optional deployment identifier for multi-tenancy. Enables multiple deployments to share the same infrastructure while maintaining data isolation." required:"N"`
 	AESEncryptionSecret string   `yaml:"aes_encryption_secret" env:"AES_ENCRYPTION_SECRET" desc:"A 16, 24, or 32 byte secret key used for AES encryption of sensitive data at rest." required:"Y"`
 	Topics              []string `yaml:"topics" env:"TOPICS" envSeparator:"," desc:"Comma-separated list of topics that this Outpost instance should subscribe to for event processing." required:"N"`
-	HTTPUserAgent string `yaml:"http_user_agent" env:"HTTP_USER_AGENT" desc:"Custom HTTP User-Agent string for outgoing webhook deliveries. If unset, defaults to 'Outpost/{version}'." required:"N"`
+	HTTPUserAgent       string   `yaml:"http_user_agent" env:"HTTP_USER_AGENT" desc:"Custom HTTP User-Agent string for outgoing webhook deliveries. If unset, defaults to 'Outpost/{version}'." required:"N"`
 
 	// Infrastructure
 	Redis       RedisConfig      `yaml:"redis"`
@@ -478,9 +478,9 @@ func (c *OperatorEventsConfig) ToConfig() opevents.Config {
 }
 
 type AlertConfig struct {
-	ConsecutiveFailureCount       int    `yaml:"consecutive_failure_count" env:"ALERT_CONSECUTIVE_FAILURE_COUNT" desc:"Number of consecutive delivery failures for a destination before triggering an alert and potentially disabling it." required:"N"`
-	AutoDisableDestination        bool   `yaml:"auto_disable_destination" env:"ALERT_AUTO_DISABLE_DESTINATION" desc:"If true, automatically disables a destination after 'consecutive_failure_count' is reached." required:"N"`
-	ExhaustedRetriesWindowSeconds int    `yaml:"exhausted_retries_window_seconds" env:"ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS" desc:"Suppression window in seconds for exhausted_retries alerts. First exhaustion per destination emits an alert; subsequent exhaustions within the window are suppressed." required:"N"`
+	ConsecutiveFailureCount       int  `yaml:"consecutive_failure_count" env:"ALERT_CONSECUTIVE_FAILURE_COUNT" desc:"Number of consecutive delivery failures for a destination before triggering an alert and potentially disabling it." required:"N"`
+	AutoDisableDestination        bool `yaml:"auto_disable_destination" env:"ALERT_AUTO_DISABLE_DESTINATION" desc:"If true, automatically disables a destination after 'consecutive_failure_count' is reached." required:"N"`
+	ExhaustedRetriesWindowSeconds int  `yaml:"exhausted_retries_window_seconds" env:"ALERT_EXHAUSTED_RETRIES_WINDOW_SECONDS" desc:"Suppression window in seconds for exhausted_retries alerts. First exhaustion per destination emits an alert; subsequent exhaustions within the window are suppressed." required:"N"`
 }
 
 // ConfigFilePath returns the path of the config file that was used

internal/logstore/pglogstore/pglogstore.go

@@ -455,12 +455,12 @@ func scanAttemptRecords(rows pgx.Rows) ([]attemptRecordWithPosition, error) {
 			attemptTime      time.Time
 			attemptNumber    int
 			manual           bool
-			code                string
-			responseDataStr     string
-			eventTime           time.Time
-			eligibleForRetry    bool
-			eventData           string
-			eventMetadata       map[string]string
+			code             string
+			responseDataStr  string
+			eventTime        time.Time
+			eligibleForRetry bool
+			eventData        string
+			eventMetadata    map[string]string
 		)
 
 		if err := rows.Scan(
@@ -642,12 +642,12 @@ func (s *logStore) RetrieveAttempt(ctx context.Context, req driver.RetrieveAttem
 		attemptTime      time.Time
 		attemptNumber    int
 		manual           bool
-		code              string
-		responseDataStr   string
-		eventTime         time.Time
-		eligibleForRetry  bool
-		eventData         string
-		eventMetadata     map[string]string
+		code             string
+		responseDataStr  string
+		eventTime        time.Time
+		eligibleForRetry bool
+		eventData        string
+		eventMetadata    map[string]string
 	)
 
 	err := row.Scan(

internal/migrator/coordinator/coordinator.go

@@ -38,11 +38,11 @@ type Coordinator struct {
 // are expected to own the lifecycle of the SQL migrator and the Redis
 // client and close them when finished.
 type Config struct {
-	SQLMigrator      *migrator.Migrator
-	RedisClient      redis.Client
-	RedisMigrations  []migratorredis.Migration
-	DeploymentID     string
-	Logger           *logging.Logger
+	SQLMigrator     *migrator.Migrator
+	RedisClient     redis.Client
+	RedisMigrations []migratorredis.Migration
+	DeploymentID    string
+	Logger          *logging.Logger
 }
 
 // New constructs a Coordinator. Either SQLMigrator or the Redis fields

internal/migrator/coordinator/coordinator_test.go

@@ -19,7 +19,7 @@ type redisTestClient struct {
 	*r.Client
 }
 
-func (c *redisTestClient) Close() error          { return c.Client.Close() }
+func (c *redisTestClient) Close() error              { return c.Client.Close() }
 func (c *redisTestClient) Pipeline() redis.Pipeliner { return c.Client.Pipeline() }
 
 // fakeMigration is a test-only migratorredis.Migration that records calls.
@@ -92,7 +92,7 @@ func (m *fakeMigration) Verify(ctx context.Context, state *migratorredis.State)
 	}, nil
 }
 
-func (m *fakeMigration) PlanCleanup(ctx context.Context) (int, error)    { return 0, nil }
+func (m *fakeMigration) PlanCleanup(ctx context.Context) (int, error) { return 0, nil }
 func (m *fakeMigration) Cleanup(ctx context.Context, state *migratorredis.State) error {
 	return nil
 }

internal/opevents/config.go

@@ -3,7 +3,9 @@ package opevents
 import (
 	"fmt"
 
+	"github.com/hookdeck/outpost/internal/logging"
 	"github.com/hookdeck/outpost/internal/mqs"
+	"go.uber.org/zap"
 )
 
 // Config holds the configuration for the operator events system.
@@ -44,9 +46,9 @@ type RabbitMQSinkConfig struct {
 
 // NewSink returns the appropriate Sink based on config.
 // Returns NoopSink if no sink is configured.
-// Returns an error if topics are specified but no sink is configured, as events
-// would be silently dropped.
-func NewSink(cfg Config) (Sink, error) {
+// If topics are specified but no sink is configured, it logs a warning and
+// returns NoopSink (operator events will be dropped).
+func NewSink(cfg Config, logger *logging.Logger) (Sink, error) {
 	if cfg.HTTP != nil {
 		return NewHTTPSink(cfg.HTTP.URL, cfg.HTTP.SigningSecret), nil
 	}
@@ -60,7 +62,8 @@ func NewSink(cfg Config) (Sink, error) {
 		return newMQSinkFromRabbitMQ(cfg.RabbitMQ)
 	}
 	if len(cfg.Topics) > 0 {
-		return nil, fmt.Errorf("opevents: topics are configured (%v) but no sink is set; events would be silently dropped", cfg.Topics)
+		logger.Warn("opevents: topics are configured but no sink is set; operator events will be dropped", zap.Any("topics", cfg.Topics))
+		return &NoopSink{}, nil
 	}
 	return &NoopSink{}, nil
 }

internal/opevents/sink_noop.go

@@ -5,6 +5,6 @@ import "context"
 // NoopSink is a sink that discards all events. Used when no sink is configured.
 type NoopSink struct{}
 
-func (s *NoopSink) Init(ctx context.Context) error                        { return nil }
-func (s *NoopSink) Send(ctx context.Context, event *OperatorEvent) error  { return nil }
-func (s *NoopSink) Close() error                                          { return nil }
+func (s *NoopSink) Init(ctx context.Context) error                       { return nil }
+func (s *NoopSink) Send(ctx context.Context, event *OperatorEvent) error { return nil }
+func (s *NoopSink) Close() error                                         { return nil }

internal/services/builder.go

@@ -193,7 +193,7 @@ func (b *ServiceBuilder) BuildAPIWorkers(baseRouter *gin.Engine) error {
 
 	// Create operator events emitter for subscription updates
 	oeCfg := b.cfg.OperatorEvents.ToConfig()
-	oeSink, err := opevents.NewSink(oeCfg)
+	oeSink, err := opevents.NewSink(oeCfg, b.logger)
 	if err != nil {
 		return fmt.Errorf("failed to create operator events sink: %w", err)
 	}
@@ -345,7 +345,7 @@ func (b *ServiceBuilder) BuildLogWorker(baseRouter *gin.Engine) error {
 
 	// Initialize alert monitor for operator events
 	oeCfg := b.cfg.OperatorEvents.ToConfig()
-	sink, err := opevents.NewSink(oeCfg)
+	sink, err := opevents.NewSink(oeCfg, b.logger)
 	if err != nil {
 		return fmt.Errorf("failed to create operator events sink: %w", err)
 	}