feat(position-keeping): integrate event outbox worker for transactional event publishing (#420)

Author: bjcoombs

services/position-keeping/app/container.go

@@ -12,6 +12,7 @@ import (
 	"github.com/meridianhub/meridian/services/position-keeping/domain"
 	"github.com/meridianhub/meridian/shared/platform/audit"
 	"github.com/meridianhub/meridian/shared/platform/auth"
+	"github.com/meridianhub/meridian/shared/platform/events"
 	"github.com/meridianhub/meridian/shared/platform/kafka"
 	"github.com/meridianhub/meridian/shared/platform/observability"
 	"github.com/redis/go-redis/v9"
@@ -35,6 +36,9 @@ type Container struct {
 
 	// Repository
 	PositionLogRepository domain.FinancialPositionLogRepository
+
+	// Event Outbox
+	OutboxRepository *events.PgxOutboxRepository
 }
 
 // NewContainer creates and initializes a new dependency injection container
@@ -67,6 +71,8 @@ func NewContainer(ctx context.Context, config *Config, logger *slog.Logger) (*Co
 
 	container.initializeRepositories()
 
+	container.initializeOutboxRepository()
+
 	logger.Info("dependency container initialized successfully")
 
 	return container, nil
@@ -310,6 +316,24 @@ func (c *Container) initializeRepositories() {
 	c.Logger.Info("repositories initialized")
 }
 
+// initializeOutboxRepository initializes the event outbox repository for transactional publishing.
+// The repository is always initialized regardless of Kafka availability because:
+// 1. Domain services use it transactionally to persist events alongside state changes
+// 2. When Kafka is disabled, events remain in the outbox until Kafka becomes available
+// 3. This enables graceful degradation - the system continues operating without message broker
+func (c *Container) initializeOutboxRepository() {
+	// TODO(tm:bian-alignment.14): Consider exposing outbox depth as a health check metric
+	// to alert operators when the outbox is backing up (e.g., Kafka unavailable).
+	c.OutboxRepository = events.NewPgxOutboxRepository(c.DBPool)
+	c.Logger.Info("event outbox repository initialized")
+}
+
+// KafkaProducer returns the Kafka producer for use by components that need
+// direct Kafka access (e.g., the event outbox worker). Returns nil if Kafka is disabled.
+func (c *Container) KafkaProducer() *kafka.ProtoProducer {
+	return c.kafkaProducer
+}
+
 // Close gracefully closes all resources in the container
 func (c *Container) Close(ctx context.Context) error {
 	c.Logger.Info("closing container resources...")

services/position-keeping/cmd/main.go

@@ -21,6 +21,7 @@ import (
 	"github.com/meridianhub/meridian/shared/pkg/idempotency"
 	"github.com/meridianhub/meridian/shared/pkg/interceptors"
 	"github.com/meridianhub/meridian/shared/platform/auth"
+	"github.com/meridianhub/meridian/shared/platform/events"
 	"github.com/prometheus/client_golang/prometheus"
 	"github.com/prometheus/client_golang/prometheus/promhttp"
 	"google.golang.org/grpc"
@@ -121,6 +122,29 @@ func run(logger *slog.Logger) error {
 
 	logger.Info("dependency container initialized")
 
+	// Initialize and start event outbox worker (if Kafka enabled)
+	// TODO(tm:bian-alignment.14): Make worker config values (batch_size, poll_interval, max_retries)
+	// configurable via environment variables for production tuning.
+	var outboxWorker *events.Worker
+	var workerCancel context.CancelFunc
+	if container.KafkaProducer() != nil {
+		workerConfig := events.DefaultWorkerConfig("position-keeping")
+		outboxWorker = events.NewWorker(
+			container.OutboxRepository,
+			container.KafkaProducer(),
+			workerConfig,
+			logger,
+		)
+
+		// Start worker in background
+		var workerCtx context.Context
+		workerCtx, workerCancel = context.WithCancel(context.Background())
+		defer workerCancel() // Safety net; primary shutdown goes through explicit cancellation
+		outboxWorker.Start(workerCtx)
+	} else {
+		logger.Info("event outbox worker disabled (kafka not configured)")
+	}
+
 	// Create idempotency service
 	var idempotencySvc idempotency.Service
 	if container.RedisClient != nil {
@@ -274,6 +298,18 @@ func run(logger *slog.Logger) error {
 	// Graceful shutdown
 	logger.Info("shutting down servers...")
 
+	// Shutdown outbox worker before stopping servers
+	// TODO(tm:bian-alignment.14): Add a shutdown timeout mechanism to prevent indefinite blocking
+	// if the worker fails to stop gracefully (e.g., Kafka broker unreachable).
+	if outboxWorker != nil {
+		logger.Info("stopping event outbox worker...")
+		if workerCancel != nil {
+			workerCancel() // Cancel context first to signal worker to stop accepting new work
+		}
+		outboxWorker.Stop() // Blocks until current batch completes and Kafka flush finishes
+		logger.Info("event outbox worker stopped")
+	}
+
 	// Create shutdown context with timeout
 	shutdownCtx, cancel := context.WithTimeout(context.Background(), config.Server.GracefulShutdownTimeout)
 	defer cancel()

shared/platform/events/worker.go

@@ -9,6 +9,7 @@ import (
 	"time"
 
 	"github.com/confluentinc/confluent-kafka-go/v2/kafka"
+	"github.com/google/uuid"
 )
 
 // Worker errors.
@@ -49,6 +50,26 @@ type KafkaPublisher interface {
 	Close()
 }
 
+// WorkerRepository defines the interface for repository operations used by the Worker.
+// This is a subset of OutboxRepository that excludes Insert, allowing both GORM-based
+// (PostgresOutboxRepository) and pgx-based (PgxOutboxRepository) implementations to be used.
+type WorkerRepository interface {
+	// FetchAndLockForProcessing atomically fetches pending entries and marks them as processing.
+	FetchAndLockForProcessing(ctx context.Context, serviceName string, limit int) ([]EventOutbox, error)
+
+	// MarkCompleted marks an entry as successfully processed.
+	MarkCompleted(ctx context.Context, id uuid.UUID) error
+
+	// MarkFailed increments retry count and updates error message.
+	MarkFailed(ctx context.Context, id uuid.UUID, err error, maxRetries int) error
+
+	// GetPendingCount returns the number of pending entries for observability.
+	GetPendingCount(ctx context.Context, serviceName string) (int64, error)
+
+	// ResetStuckEntries resets entries stuck in 'processing' state for too long.
+	ResetStuckEntries(ctx context.Context, serviceName string, olderThan time.Duration) (int64, error)
+}
+
 // WorkerConfig contains configuration for the event outbox worker.
 type WorkerConfig struct {
 	// ServiceName identifies this service for filtering outbox entries.
@@ -90,7 +111,7 @@ func DefaultWorkerConfig(serviceName string) WorkerConfig {
 // Worker is a background processor that publishes events from the outbox to Kafka.
 // It implements graceful shutdown and handles retries with exponential backoff.
 type Worker struct {
-	repository   OutboxRepository
+	repository   WorkerRepository
 	publisher    KafkaPublisher
 	config       WorkerConfig
 	logger       *slog.Logger
@@ -109,7 +130,7 @@ type Worker struct {
 //
 // Returns a configured Worker ready to start processing.
 func NewWorker(
-	repository OutboxRepository,
+	repository WorkerRepository,
 	publisher KafkaPublisher,
 	config WorkerConfig,
 	logger *slog.Logger,

shared/platform/kafka/producer.go

@@ -170,6 +170,21 @@ func (p *ProtoProducer) Close() {
 	p.producer.Close()
 }
 
+// Produce sends a raw Kafka message asynchronously with delivery report.
+// This is a low-level method that exposes the underlying Kafka producer's Produce method,
+// allowing callers to receive delivery confirmations via the deliveryChan.
+// This method is used by the event outbox worker to publish pre-serialized event payloads.
+//
+// Parameters:
+// - msg: Pre-built Kafka message with topic, key, value, and optional headers
+// - deliveryChan: Channel to receive delivery confirmation events
+//
+// Returns an error if the message cannot be enqueued for delivery.
+// Note: A nil return does not guarantee delivery - check the deliveryChan for confirmation.
+func (p *ProtoProducer) Produce(msg *kafka.Message, deliveryChan chan kafka.Event) error {
+	return p.producer.Produce(msg, deliveryChan)
+}
+
 // PublishWithTenant sends a protobuf message with tenant context to the specified Kafka topic.
 // The tenant ID is extracted from the context and injected as a Kafka header (x-tenant-id).
 // This ensures tenant isolation for multi-tenant event processing.