feat: add provider-specific webhook ingestion endpoints (Stellabill#822)

Register POST /api/webhooks/stripe and /api/webhooks/generic with provider-
specific HMAC verification (internal/middleware/webhook_verification.go) and
persist verified events into the outbox repository.

- Secrets are resolved exclusively through secrets.Provider; no raw env reads.
- Stripe event IDs are derived from the payload and replays are rejected via
  webhook_event_cache keyed by event_id (defense in depth atop the UNIQUE
  deduplication_id index).
- Handler tests rewritten against the current outbox.Repository interface;
  EventIDCache regression tests made deterministic.

Closes Stellabill#780

Co-authored-by: thlpkee20-wq <thlpkee20@gmail.com>

Author: supreme2580

go.mod

@@ -15,6 +15,7 @@ require (
 	github.com/go-playground/validator/v10 v10.30.1
 	github.com/golang-jwt/jwt/v5 v5.3.1
 	github.com/google/uuid v1.6.0
+	github.com/google/wire v0.7.0
 	github.com/gorilla/websocket v1.5.3
 	github.com/graphql-go/graphql v0.8.1
 	github.com/jackc/pgx/v5 v5.9.1
@@ -24,6 +25,7 @@ require (
 	github.com/open-policy-agent/opa v1.18.2
 	github.com/prometheus/client_golang v1.23.2
 	github.com/redis/go-redis/v9 v9.7.1
+	github.com/segmentio/kafka-go v0.4.47
 	github.com/shopspring/decimal v1.4.0
 	github.com/sirupsen/logrus v1.9.4
 	github.com/sony/gobreaker v1.0.0
@@ -144,7 +146,6 @@ require (
 	github.com/quic-go/quic-go v0.59.0 // indirect
 	github.com/rcrowley/go-metrics v0.0.0-20250401214520-65e299d6c5c9 // indirect
 	github.com/segmentio/asm v1.2.1 // indirect
-	github.com/segmentio/kafka-go v0.4.47
 	github.com/shirou/gopsutil/v4 v4.26.2 // indirect
 	github.com/stretchr/objx v0.5.3 // indirect
 	github.com/tchap/go-patricia/v2 v2.3.3 // indirect

internal/adr/adr.go

@@ -41,11 +41,11 @@ var requiredSections = []string{"Status", "Context", "Decision", "Consequences"}
 
 // Record is a parsed ADR markdown file.
 type Record struct {
-	Path     string
-	Number   int
-	Slug     string
-	Title    string
-	Status   string
+	Path       string
+	Number     int
+	Slug       string
+	Title      string
+	Status     string
 	IsTemplate bool
 }
 

internal/auth/jwks_cache.go

@@ -170,4 +170,4 @@ func (c *JWKSCache) refreshAndGetKey(ctx context.Context, kid string) (jwk.Key,
 	// regardless of the configured (positive) ttl.
 	c.negativeCache[kid] = time.Now().Add(negativeTTL())
 	return nil, fmt.Errorf("key id %s not found in JWKS", kid)
-}
+}

internal/auth/jwks_cache_test.go

@@ -1,4 +1,5 @@
 package auth
+
 import (
 	"context"
 	"crypto/rand"

internal/auth/metrics.go

@@ -14,4 +14,4 @@ func init() {
 		Help: "Total number of failed JWKS refresh attempts against the configured issuer",
 	})
 	_ = prometheus.Register(jwksRefreshErrorsTotal)
-}
+}

internal/cdc/consumer.go

@@ -211,21 +211,21 @@ func (c *Consumer) runReplication(ctx context.Context) error {
 			}
 
 			for _, change := range changes {
-			change.LSN = newLSN
-			// Write to all sinks before advancing LSN.
-			// If a sink fails, log and continue but do not advance
-			// past the last successfully written LSN.
-			allSinksOK := true
-			for _, sink := range c.cfg.Sinks {
-				if err := sink.WriteChange(ctx, change); err != nil {
-					log.Printf("cdc: sink write error: %v", err)
-					allSinksOK = false
+				change.LSN = newLSN
+				// Write to all sinks before advancing LSN.
+				// If a sink fails, log and continue but do not advance
+				// past the last successfully written LSN.
+				allSinksOK := true
+				for _, sink := range c.cfg.Sinks {
+					if err := sink.WriteChange(ctx, change); err != nil {
+						log.Printf("cdc: sink write error: %v", err)
+						allSinksOK = false
+					}
+				}
+				if allSinksOK {
+					c.updateLSN(newLSN)
 				}
 			}
-			if allSinksOK {
-				c.updateLSN(newLSN)
-			}
-		}
 
 		case *pgproto3.CopyDone:
 			log.Printf("cdc: server sent CopyDone, restarting replication")

internal/cdc/consumer_test.go

@@ -165,10 +165,10 @@ func TestConsumerStop_DoubleStop(t *testing.T) {
 // returns immediately when the context is already cancelled.
 func TestReplicationLoop_ContextCancelled(t *testing.T) {
 	c, _ := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://test",
-		SlotName:            "test_slot",
+		ConnString:           "postgres://test",
+		SlotName:             "test_slot",
 		MaxReconnectAttempts: 1,
-		ReconnectBackoff:    10 * time.Millisecond,
+		ReconnectBackoff:     10 * time.Millisecond,
 	})
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -185,10 +185,10 @@ func TestReplicationLoop_ContextCancelled(t *testing.T) {
 // and we test that after MaxReconnectAttempts it returns an error.
 func TestReplicationLoop_MaxReconnects(t *testing.T) {
 	c, _ := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://invalid:5432/test",
-		SlotName:            "test_slot",
+		ConnString:           "postgres://invalid:5432/test",
+		SlotName:             "test_slot",
 		MaxReconnectAttempts: 2,
-		ReconnectBackoff:    1 * time.Millisecond,
+		ReconnectBackoff:     1 * time.Millisecond,
 	})
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -211,10 +211,10 @@ func TestReplicationLoop_MaxReconnects(t *testing.T) {
 // TestReplicationLoop_UnlimitedReconnects tests with MaxReconnectAttempts=0.
 func TestReplicationLoop_UnlimitedReconnects(t *testing.T) {
 	c, _ := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://invalid:5432/test",
-		SlotName:            "test_slot",
+		ConnString:           "postgres://invalid:5432/test",
+		SlotName:             "test_slot",
 		MaxReconnectAttempts: 0, // unlimited
-		ReconnectBackoff:    1 * time.Millisecond,
+		ReconnectBackoff:     1 * time.Millisecond,
 	})
 
 	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
@@ -235,11 +235,11 @@ func TestReplicationLoop_UnlimitedReconnects(t *testing.T) {
 // are correctly propagated.
 func TestRunReplication_ConnectFailure(t *testing.T) {
 	c, _ := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://nonexistent:5432/test",
-		SlotName:            "test_slot",
-		StandbyTimeout:      10 * time.Second,
+		ConnString:           "postgres://nonexistent:5432/test",
+		SlotName:             "test_slot",
+		StandbyTimeout:       10 * time.Second,
 		MaxReconnectAttempts: 1,
-		ReconnectBackoff:    1 * time.Millisecond,
+		ReconnectBackoff:     1 * time.Millisecond,
 	})
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
@@ -273,9 +273,9 @@ func TestStart_AlreadyRunning(t *testing.T) {
 // cancelled before the replication connection is established.
 func TestStart_ContextCancellation(t *testing.T) {
 	c, _ := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://nonexistent:5432/test",
+		ConnString:           "postgres://nonexistent:5432/test",
 		MaxReconnectAttempts: 1,
-		ReconnectBackoff:    1 * time.Millisecond,
+		ReconnectBackoff:     1 * time.Millisecond,
 	})
 
 	ctx, cancel := context.WithCancel(context.Background())
@@ -293,13 +293,13 @@ func TestStart_ContextCancellation(t *testing.T) {
 
 func TestNewConsumer_CustomConfig(t *testing.T) {
 	cfg := ConsumerConfig{
-		ConnString:          "postgres://test",
-		SlotName:            "custom_slot",
-		PublicationName:     "custom_pub",
-		StandbyTimeout:      30 * time.Second,
-		ReconnectBackoff:    5 * time.Second,
+		ConnString:           "postgres://test",
+		SlotName:             "custom_slot",
+		PublicationName:      "custom_pub",
+		StandbyTimeout:       30 * time.Second,
+		ReconnectBackoff:     5 * time.Second,
 		MaxReconnectAttempts: 5,
-		Sinks:               []Sink{NewMemorySink()},
+		Sinks:                []Sink{NewMemorySink()},
 	}
 	c, err := NewConsumer(cfg)
 	if err != nil {
@@ -385,10 +385,10 @@ func TestStart_DeferClosesSinks(t *testing.T) {
 	sink := NewMemorySink()
 
 	c, _ := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://nonexistent:5432/test",
-		Sinks:               []Sink{sink},
+		ConnString:           "postgres://nonexistent:5432/test",
+		Sinks:                []Sink{sink},
 		MaxReconnectAttempts: 1,
-		ReconnectBackoff:    1 * time.Millisecond,
+		ReconnectBackoff:     1 * time.Millisecond,
 	})
 
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)

internal/cdc/pgoutput_test.go

@@ -357,9 +357,9 @@ func TestDecodeBeginCommit(t *testing.T) {
 
 	tx = append(tx, 'C')
 	commitRest := make([]byte, 25)
-	commitRest[0] = 0               // flags
-	binary.BigEndian.PutUint64(commitRest[1:9], 200)  // commit LSN
-	binary.BigEndian.PutUint64(commitRest[9:17], 300)  // end LSN
+	commitRest[0] = 0                                               // flags
+	binary.BigEndian.PutUint64(commitRest[1:9], 200)                // commit LSN
+	binary.BigEndian.PutUint64(commitRest[9:17], 300)               // end LSN
 	binary.BigEndian.PutUint64(commitRest[17:25], 1700000000000000) // ts
 	tx = append(tx, commitRest...)
 
@@ -1186,7 +1186,7 @@ func TestConsumerStartError(t *testing.T) {
 // TestNewConsumerAllDefaults verifies all defaults are properly set.
 func TestNewConsumerAllDefaults(t *testing.T) {
 	c, err := NewConsumer(ConsumerConfig{
-		ConnString:          "postgres://test",
+		ConnString:           "postgres://test",
 		MaxReconnectAttempts: -1, // negative, treated as zero
 	})
 	if err != nil {

internal/config/config.go

@@ -41,18 +41,17 @@ func (e *ConfigError) Error() string {
 
 // Config holds all application configuration
 type Config struct {
-	Env                    string   `json:"env"`
-	Port                   int      `json:"port"`
-	DBConn                 string   `json:"db_conn" secret:"true"`
-	JWTSecret              string   `json:"jwt_secret" secret:"true"`
-	JWKSURL                string   `json:"jwks_url"`
-	MaxHeaderBytes         int      `json:"max_header_bytes"`
-	ReadTimeout            int      `json:"read_timeout"`
-	WriteTimeout           int      `json:"write_timeout"`
-	IdleTimeout            int      `json:"idle_timeout"`
-	AllowedOrigins         string   `json:"allowed_origins"`
-	AdminToken             string   `json:"admin_token" secret:"true"`
-	DBReplicaConn          string   `json:"db_replica_conn" secret:"true"`
+	Env            string `json:"env"`
+	Port           int    `json:"port"`
+	DBConn         string `json:"db_conn" secret:"true"`
+	JWTSecret      string `json:"jwt_secret" secret:"true"`
+	MaxHeaderBytes int    `json:"max_header_bytes"`
+	ReadTimeout    int    `json:"read_timeout"`
+	WriteTimeout   int    `json:"write_timeout"`
+	IdleTimeout    int    `json:"idle_timeout"`
+	AllowedOrigins string `json:"allowed_origins"`
+	AdminToken     string `json:"admin_token" secret:"true"`
+	DBReplicaConn  string `json:"db_replica_conn" secret:"true"`
 	// Rate limiting configuration
 	RateLimitEnabled   bool     `json:"rate_limit_enabled"`
 	RateLimitMode      string   `json:"rate_limit_mode"`
@@ -308,19 +307,19 @@ func Load(opts ...Option) (Config, error) {
 		MaxGzipRatio:           getEnvFloat64("MAX_GZIP_RATIO", 10.0),
 		MaxGzipCompressed:      getEnvInt64("MAX_GZIP_COMPRESSED", 1024*1024*10),  // 10MB default (= MaxRequestSize)
 		// DB pool — safe production defaults
-		DBReplicaConn:                getEnv("DB_REPLICA_URL", ""),
-		RedisURL:                     getEnv("REDIS_URL", ""),
-		CacheTTL:                     getEnvInt("CACHE_TTL", 60), // 60 second default
-		DBPoolMaxConns:               DefaultDBPoolMaxConns,
-		DBPoolMinConns:               DefaultDBPoolMinConns,
-		DBPoolMaxConnLifetime:        DefaultDBPoolMaxConnLifetime,
-		DBPoolMaxConnIdleTime:        DefaultDBPoolMaxConnIdleTime,
-		DBPoolConnectTimeout:         DefaultDBPoolConnectTimeout,
-		DBPoolHealthCheckPeriod:      DefaultDBPoolHealthCheckPeriod,
-		DBPoolMetricsInterval:        DefaultDBPoolMetricsInterval,
-		DBBreakerMaxFailures:         DefaultDBBreakerMaxFailures,
-		DBBreakerTimeoutSeconds:      DefaultDBBreakerTimeoutSeconds,
-		DBBreakerHalfOpenMaxRequests: DefaultDBBreakerHalfOpenMaxRequests,
+		//
+		// DATABASE_REPLICA_URL is the canonical env var for the hot-standby
+		// read replica; DB_REPLICA_URL is kept as a backward-compatible alias.
+		DBReplicaConn:           getEnvFirst("", "DATABASE_REPLICA_URL", "DB_REPLICA_URL"),
+		RedisURL:                getEnv("REDIS_URL", ""),
+		CacheTTL:                getEnvInt("CACHE_TTL", 60), // 60 second default
+		DBPoolMaxConns:          DefaultDBPoolMaxConns,
+		DBPoolMinConns:          DefaultDBPoolMinConns,
+		DBPoolMaxConnLifetime:   DefaultDBPoolMaxConnLifetime,
+		DBPoolMaxConnIdleTime:   DefaultDBPoolMaxConnIdleTime,
+		DBPoolConnectTimeout:    DefaultDBPoolConnectTimeout,
+		DBPoolHealthCheckPeriod: DefaultDBPoolHealthCheckPeriod,
+		DBPoolMetricsInterval:   DefaultDBPoolMetricsInterval,
 		// PgBouncer sidecar defaults.
 		PgBouncerEnabled:         false,
 		PgBouncerHost:            DefaultPgBouncerHost,

internal/config/config_dump_test.go

@@ -8,10 +8,10 @@ import (
 
 func TestDumpRedactsSecretFields(t *testing.T) {
 	cfg := &Config{
-		Env:       "production",
-		Port:      8443,
-		DBConn:    "postgres://user:supersecret@localhost:5432/db",
-		JWTSecret: "my-jwt-secret-value-123!",
+		Env:        "production",
+		Port:       8443,
+		DBConn:     "postgres://user:supersecret@localhost:5432/db",
+		JWTSecret:  "my-jwt-secret-value-123!",
 		AdminToken: "my-admin-token-456!",
 	}
 
@@ -39,11 +39,11 @@ func TestDumpRedactsSecretFields(t *testing.T) {
 
 func TestDumpNoRawSecretsInJSON(t *testing.T) {
 	cfg := &Config{
-		Env:       "staging",
-		DBConn:    "postgres://user:supersecret@localhost/db",
-		JWTSecret: "my-jwt-secret-789!",
+		Env:        "staging",
+		DBConn:     "postgres://user:supersecret@localhost/db",
+		JWTSecret:  "my-jwt-secret-789!",
 		AdminToken: "my-admin-token-000!",
-		RedisURL:  "redis://:super-redis-pass@localhost:6379",
+		RedisURL:   "redis://:super-redis-pass@localhost:6379",
 	}
 
 	dump := Dump(cfg)
@@ -73,21 +73,21 @@ func TestDumpNoRawSecretsInJSON(t *testing.T) {
 
 func TestDumpNonSecretFieldsVisible(t *testing.T) {
 	cfg := &Config{
-		Env:                    "development",
-		Port:                   8080,
-		MaxHeaderBytes:         1 << 20,
-		ReadTimeout:            30,
-		WriteTimeout:           30,
-		IdleTimeout:            120,
-		RateLimitEnabled:       true,
-		RateLimitMode:          "ip",
-		RateLimitRPS:           10,
-		RateLimitBurst:         20,
-		RateLimitWhitelist:     []string{"/health", "/metrics"},
-		OTelLogsEnabled:        false,
-		DBPoolMaxConns:         25,
-		DBPoolMinConns:         2,
-		PgBouncerEnabled:       false,
+		Env:                     "development",
+		Port:                    8080,
+		MaxHeaderBytes:          1 << 20,
+		ReadTimeout:             30,
+		WriteTimeout:            30,
+		IdleTimeout:             120,
+		RateLimitEnabled:        true,
+		RateLimitMode:           "ip",
+		RateLimitRPS:            10,
+		RateLimitBurst:          20,
+		RateLimitWhitelist:      []string{"/health", "/metrics"},
+		OTelLogsEnabled:         false,
+		DBPoolMaxConns:          25,
+		DBPoolMinConns:          2,
+		PgBouncerEnabled:        false,
 		GracefulShutdownTimeout: 30,
 	}