refactor(metadata): replace PostgreSQL DLQ with Watermill Kafka DLQ

batazor · batazor · commit ae67b6306c29 · 2025-11-27T17:46:09.000+03:00
- Remove PostgreSQL DLQ implementation (repository/dlq, migrations, archive.go)
- Enable Watermill built-in Kafka DLQ (WATERMILL_DLQ_ENABLED=true)
- Simplify subscribe.go: remove custom DLQ logic, rely on Watermill retry/DLQ
- Remove STORE_POSTGRES_URI secret from metadata Helm values
- Update wire.go: remove NewDLQStore and rawEvents parameter
- Remove ADR-0042 (PostgreSQL DLQ is no longer used)
diff --git a/boundaries/common/ops/common/templates/store/configmap.yaml b/boundaries/common/ops/common/templates/store/configmap.yaml
@@ -9,27 +9,42 @@ data:
     -- link database
     \c link;
     CREATE SCHEMA AUTHORIZATION link;
-    GRANT ALL PRIVILEGES ON DATABASE link TO link;
     ALTER DATABASE link OWNER TO link;
 
     GRANT CONNECT ON DATABASE link TO grafana;
     GRANT USAGE ON SCHEMA link TO grafana;
-    ALTER DEFAULT PRIVILEGES IN SCHEMA link GRANT SELECT ON TABLES TO grafana;
+    ALTER DEFAULT PRIVILEGES FOR ROLE link IN SCHEMA link GRANT SELECT ON TABLES TO grafana;
+    ALTER DEFAULT PRIVILEGES FOR ROLE link IN SCHEMA link GRANT USAGE ON SEQUENCES TO grafana;
 
+    -- statement_timeout applies globally to all connections (including Grafana and maintenance)
     ALTER DATABASE link SET statement_timeout = '180s';
 
     -- proxy database
     \c proxy;
     CREATE SCHEMA AUTHORIZATION proxy;
-    GRANT ALL PRIVILEGES ON DATABASE proxy TO proxy;
     ALTER DATABASE proxy OWNER TO proxy;
 
     GRANT CONNECT ON DATABASE proxy TO grafana;
     GRANT USAGE ON SCHEMA proxy TO grafana;
-    ALTER DEFAULT PRIVILEGES IN SCHEMA proxy GRANT SELECT ON TABLES TO grafana;
+    ALTER DEFAULT PRIVILEGES FOR ROLE proxy IN SCHEMA proxy GRANT SELECT ON TABLES TO grafana;
+    ALTER DEFAULT PRIVILEGES FOR ROLE proxy IN SCHEMA proxy GRANT USAGE ON SEQUENCES TO grafana;
 
+    -- statement_timeout applies globally to all connections (including Grafana and maintenance)
     ALTER DATABASE proxy SET statement_timeout = '180s';
 
+    -- metadata database
+    \c metadata;
+    CREATE SCHEMA AUTHORIZATION metadata;
+    ALTER DATABASE metadata OWNER TO metadata;
+
+    GRANT CONNECT ON DATABASE metadata TO grafana;
+    GRANT USAGE ON SCHEMA metadata TO grafana;
+    ALTER DEFAULT PRIVILEGES FOR ROLE metadata IN SCHEMA metadata GRANT SELECT ON TABLES TO grafana;
+    ALTER DEFAULT PRIVILEGES FOR ROLE metadata IN SCHEMA metadata GRANT USAGE ON SEQUENCES TO grafana;
+
+    -- statement_timeout applies globally to all connections (including Grafana and maintenance)
+    ALTER DATABASE metadata SET statement_timeout = '180s';
+
     -- Grant grafana access to pg_statio_user_tables via pg_read_all_stats role
     GRANT pg_read_all_stats TO grafana;
     GRANT pg_read_all_settings TO grafana;
diff --git a/boundaries/common/ops/common/templates/store/postgres.yaml b/boundaries/common/ops/common/templates/store/postgres.yaml
@@ -98,12 +98,12 @@ spec:
           max_parallel_workers: 8
           max_parallel_maintenance_workers: 4
           # Synchronous Replication
-          synchronous_commit: 'on'
+          synchronous_commit: "on"
           # Partitioning
-          enable_partition_pruning: 'on'
+          enable_partition_pruning: "on"
           # TODO: enabled pgrouting,timescaledb,postgis,pgmonitor_bgw
           shared_preload_libraries: pg_stat_statements,auto_explain,pgaudit,pg_prewarm,uuid-ossp,pg_partman_bgw
-          pgmonitor_bgw.dbname: postgres,link
+          pgmonitor_bgw.dbname: postgres,link,metadata
           pgmonitor_bgw.role: "postgres"
     syncPeriodSeconds: 60
     leaderLeaseDurationSeconds: 300
@@ -130,3 +130,7 @@ spec:
     - name: proxy
       databases:
         - proxy
+    - name: metadata
+      databases:
+        - metadata
+      options: "SUPERUSER"
diff --git a/boundaries/link/ops/link/values.yaml b/boundaries/link/ops/link/values.yaml
@@ -70,7 +70,7 @@ deploy:
     WATERMILL_KAFKA_BROKERS: shortlink-kafka-bootstrap.kafka.svc.cluster.local:9092
     WATERMILL_KAFKA_CLIENT_ID: shortlink-link
     WATERMILL_KAFKA_CONSUMER_GROUP: shortlink-link
-    WATERMILL_DLQ_ENABLED: false
+    WATERMILL_DLQ_ENABLED: true
     WATERMILL_DLQ_MAX_RETRIES: 5
     TRACER_URI: grafana-tempo.grafana:4317
     GRPC_CLIENT_HOST: istio-ingress.istio-ingress
diff --git a/boundaries/metadata/internal/di/wire.go b/boundaries/metadata/internal/di/wire.go
@@ -22,12 +22,12 @@ import (
 	"github.com/shortlink-org/go-sdk/flight_trace"
 	rpc "github.com/shortlink-org/go-sdk/grpc"
 	"github.com/shortlink-org/go-sdk/logger"
-	"github.com/shortlink-org/go-sdk/watermill"
-	watermill_kafka "github.com/shortlink-org/go-sdk/watermill/backends/kafka"
 	"github.com/shortlink-org/go-sdk/observability/metrics"
 	"github.com/shortlink-org/go-sdk/observability/profiling"
 	"github.com/shortlink-org/go-sdk/observability/tracing"
 	"github.com/shortlink-org/go-sdk/s3"
+	"github.com/shortlink-org/go-sdk/watermill"
+	watermill_kafka "github.com/shortlink-org/go-sdk/watermill/backends/kafka"
 	"go.opentelemetry.io/otel/metric"
 	api "go.opentelemetry.io/otel/sdk/metric"
 	"go.opentelemetry.io/otel/trace"
diff --git a/boundaries/metadata/internal/infrastructure/mq/mq.go b/boundaries/metadata/internal/infrastructure/mq/mq.go
@@ -13,13 +13,11 @@ import (
 type Event struct {
 	subscriber message.Subscriber
 	metadataUC *metadata_uc.UC
-	rawEvents  RawEventsStore
 }
 
 func New(subscriber message.Subscriber, metadataUC *metadata_uc.UC) (*Event, error) {
 	return &Event{
 		subscriber: subscriber,
 		metadataUC: metadataUC,
-		rawEvents:  newInMemoryRawEventsStore(),
 	}, nil
 }
diff --git a/boundaries/metadata/internal/infrastructure/mq/raw_events_store.go b/boundaries/metadata/internal/infrastructure/mq/raw_events_store.go
diff --git a/boundaries/metadata/internal/infrastructure/mq/subscribe.go b/boundaries/metadata/internal/infrastructure/mq/subscribe.go
@@ -7,7 +7,6 @@ import (
 	"log/slog"
 
 	linkpb "buf.build/gen/go/shortlink-org/shortlink-link-link/protocolbuffers/go/domain/link/v1"
-	"github.com/ThreeDotsLabs/watermill/message"
 	"github.com/shortlink-org/go-sdk/cqrs/bus"
 	cqrsmessage "github.com/shortlink-org/go-sdk/cqrs/message"
 	"github.com/shortlink-org/go-sdk/logger"
@@ -39,30 +38,51 @@ func (e *Event) SubscribeLinkCreated(ctx context.Context, log logger.Logger, reg
 	}
 
 	go func(ctx context.Context) {
+		defer func() {
+			if r := recover(); r != nil {
+				log.ErrorWithContext(ctx, "panic in LinkCreated subscriber",
+					slog.Any("recover", r),
+					slog.String("topic", linkCreatedEvent),
+				)
+			}
+		}()
+
 		for msg := range messages {
 			msgCtx := msg.Context() //nolint:contextcheck // inherit context from Watermill message
 			if msgCtx == nil {
 				msgCtx = ctx
 			}
 
+			// Validate payload before unmarshaling
+			if len(msg.Payload) == 0 {
+				log.ErrorWithContext(msgCtx, "Received empty payload for link created event - nacking for Kafka DLQ",
+					slog.String("topic", linkCreatedEvent),
+					slog.String("message_uuid", msg.UUID),
+				)
+
+				msg.Nack()
+				continue
+			}
+
 			// Create typed event instance directly - no reflect.New needed
 			// We know the type is *linkpb.LinkCreated from the subscription
 			event := &linkpb.LinkCreated{}
 
 			// Unmarshal using ProtoMarshaler (handles metadata extraction)
-			watermillMsg := message.NewMessage(msg.UUID, msg.Payload)
-			watermillMsg.Metadata = msg.Metadata
-			watermillMsg.SetContext(msgCtx) //nolint:contextcheck // watermill message already carries derived context
+			// msg is already *message.Message from Watermill, just update context if needed
+			msg.SetContext(msgCtx) //nolint:contextcheck // update context for unmarshaling
 
-			unmarshalErr := marshaler.Unmarshal(watermillMsg, event)
+			unmarshalErr := marshaler.Unmarshal(msg, event)
 			if unmarshalErr != nil {
-				log.ErrorWithContext(msgCtx, "Failed to unmarshal event using marshaler", //nolint:contextcheck // logging must use message context
+				log.ErrorWithContext(msgCtx, "Failed to unmarshal event using marshaler - nacking for Kafka DLQ",
 					slog.String("error", unmarshalErr.Error()),
 					slog.String("topic", linkCreatedEvent),
+					slog.Int("payload_size", len(msg.Payload)),
+					slog.Int("metadata_count", len(msg.Metadata)),
+					slog.String("message_uuid", msg.UUID),
 				)
-				e.archiveRawEvent(msgCtx, msg, linkCreatedEvent, fmt.Sprintf("unmarshal: %v", unmarshalErr))
-				msg.Nack()
 
+				msg.Nack()
 				continue
 			}
 
@@ -72,19 +92,18 @@ func (e *Event) SubscribeLinkCreated(ctx context.Context, log logger.Logger, reg
 				var domainErr *domainerrors.Error
 				if errors.As(handleErr, &domainErr) {
 					dto := infraerrors.FromDomainError("metadata.mq.link_created", domainErr)
-					log.ErrorWithContext(msgCtx, "Failed to handle link created event",
+					log.ErrorWithContext(msgCtx, "Failed to handle link created event - nacking for Kafka DLQ",
 						slog.String("error_code", dto.Code),
 						slog.String("topic", linkCreatedEvent),
 						slog.Bool("retryable", dto.Retryable),
 						slog.String("message", dto.Message),
 					)
-					e.archiveRawEvent(msgCtx, msg, linkCreatedEvent, fmt.Sprintf("domain: %s", dto.Message))
 				} else {
-					log.ErrorWithContext(msgCtx, "Failed to handle link created event",
+					log.ErrorWithContext(msgCtx, "Failed to handle link created event - nacking for Kafka DLQ",
 						slog.String("error", handleErr.Error()),
 						slog.String("topic", linkCreatedEvent),
+						slog.Bool("retryable", true),
 					)
-					e.archiveRawEvent(msgCtx, msg, linkCreatedEvent, fmt.Sprintf("handler: %v", handleErr))
 				}
 
 				msg.Nack()
@@ -145,15 +164,3 @@ func (e *Event) handleLinkCreated(ctx context.Context, event *linkpb.LinkCreated
 	return nil
 }
 
-func (e *Event) archiveRawEvent(ctx context.Context, msg *message.Message, topic, reason string) {
-	if e == nil || e.rawEvents == nil || msg == nil {
-		return
-	}
-
-	_ = e.rawEvents.Save(ctx, RawEventRecord{
-		Topic:    topic,
-		Payload:  cloneBytes(msg.Payload),
-		Metadata: cloneMetadata(msg.Metadata),
-		Error:    reason,
-	})
-}
diff --git a/boundaries/metadata/ops/metadata/values.schema.json b/boundaries/metadata/ops/metadata/values.schema.json
@@ -37,6 +37,12 @@
             },
             "WATERMILL_KAFKA_CONSUMER_GROUP": {
               "type": "string"
+            },
+            "WATERMILL_DLQ_ENABLED": {
+              "type": "boolean"
+            },
+            "WATERMILL_DLQ_MAX_RETRIES": {
+              "type": "integer"
             }
           },
           "type": "object"
diff --git a/boundaries/metadata/ops/metadata/values.yaml b/boundaries/metadata/ops/metadata/values.yaml
@@ -51,6 +51,8 @@ deploy:
     WATERMILL_KAFKA_BROKERS: "shortlink-kafka-bootstrap.kafka:9092"
     WATERMILL_KAFKA_CLIENT_ID: "shortlink-metadata"
     WATERMILL_KAFKA_CONSUMER_GROUP: "shortlink-metadata"
+    WATERMILL_DLQ_ENABLED: true
+    WATERMILL_DLQ_MAX_RETRIES: 3
     TRACER_URI: grafana-tempo.grafana:4317
     GRPC_CLIENT_HOST: istio-ingress.istio-ingress
 

Original file line number	Diff line number	Diff line change
`@@ -13,13 +13,11 @@ import (`
`13`	`13`	`type Event struct {`
`14`	`14`	`subscriber message.Subscriber`
`15`	`15`	`metadataUC *metadata_uc.UC`
`16`		`- rawEvents RawEventsStore`
`17`	`16`	`}`
`18`	`17`
`19`	`18`	`func New(subscriber message.Subscriber, metadataUC metadata_uc.UC) (Event, error) {`
`20`	`19`	`return &Event{`
`21`	`20`	`subscriber: subscriber,`
`22`	`21`	`metadataUC: metadataUC,`
`23`		`- rawEvents: newInMemoryRawEventsStore(),`
`24`	`22`	`}, nil`
`25`	`23`	`}`