feat(cqrs): add outbox pattern support with pgxpool

- Add OutboxConfig with support for both *sql.DB and *pgxpool.Pool
- Implement forwarder state management for Watermill outbox/forwarder
- Add RunForwarder methods to CommandBus and EventBus
- Support automatic conversion from pgxpool.Pool to *sql.DB using stdlib
- Add logger and meter provider support for observability

Author: batazor

cqrs/bus/command_bus.go

@@ -21,15 +21,43 @@ type CommandBus struct {
 	publisher wmmessage.Publisher
 	marshaler cqrsmessage.Marshaler
 	namer     cqrsmessage.Namer
+	forwarder *forwarderState
 }
 
 // NewCommandBus builds a bus backed by Watermill publisher.
-func NewCommandBus(pub wmmessage.Publisher, marshaler cqrsmessage.Marshaler, namer cqrsmessage.Namer) *CommandBus {
-	return &CommandBus{
+func NewCommandBus(pub wmmessage.Publisher, marshaler cqrsmessage.Marshaler, namer cqrsmessage.Namer, opts ...Option) *CommandBus {
+	bus, err := NewCommandBusWithOptions(pub, marshaler, namer, opts...)
+	if err != nil {
+		panic(err)
+	}
+
+	return bus
+}
+
+// NewCommandBusWithOptions builds a bus and returns configuration errors instead of panicking.
+func NewCommandBusWithOptions(
+	pub wmmessage.Publisher,
+	marshaler cqrsmessage.Marshaler,
+	namer cqrsmessage.Namer,
+	opts ...Option,
+) (*CommandBus, error) {
+	cfg, err := applyOptions(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	bus := &CommandBus{
 		publisher: pub,
 		marshaler: marshaler,
 		namer:     namer,
 	}
+
+	if cfg.outbox != nil {
+		bus.forwarder = newForwarderState(cfg.outbox)
+		bus.publisher = bus.forwarder.wrapPublisher(pub)
+	}
+
+	return bus, nil
 }
 
 // validate checks that the command bus and its dependencies are properly initialized.
@@ -90,3 +118,21 @@ func (b *CommandBus) Send(ctx context.Context, cmd any) error {
 
 	return b.publisher.Publish(topic, msg)
 }
+
+// RunForwarder starts the optional outbox forwarder when configured.
+func (b *CommandBus) RunForwarder(ctx context.Context) error {
+	if b == nil || b.forwarder == nil {
+		return nil
+	}
+
+	return b.forwarder.Run(ctx)
+}
+
+// CloseForwarder attempts to stop previously started forwarder gracefully.
+func (b *CommandBus) CloseForwarder(ctx context.Context) error {
+	if b == nil || b.forwarder == nil {
+		return nil
+	}
+
+	return b.forwarder.Close(ctx)
+}

cqrs/bus/event_bus.go

@@ -21,15 +21,43 @@ type EventBus struct {
 	publisher wmmessage.Publisher
 	marshaler cqrsmessage.Marshaler
 	namer     cqrsmessage.Namer
+	forwarder *forwarderState
 }
 
 // NewEventBus builds EventBus with required dependencies.
-func NewEventBus(pub wmmessage.Publisher, marshaler cqrsmessage.Marshaler, namer cqrsmessage.Namer) *EventBus {
-	return &EventBus{
+func NewEventBus(pub wmmessage.Publisher, marshaler cqrsmessage.Marshaler, namer cqrsmessage.Namer, opts ...Option) *EventBus {
+	bus, err := NewEventBusWithOptions(pub, marshaler, namer, opts...)
+	if err != nil {
+		panic(err)
+	}
+
+	return bus
+}
+
+// NewEventBusWithOptions builds EventBus and exposes configuration errors.
+func NewEventBusWithOptions(
+	pub wmmessage.Publisher,
+	marshaler cqrsmessage.Marshaler,
+	namer cqrsmessage.Namer,
+	opts ...Option,
+) (*EventBus, error) {
+	cfg, err := applyOptions(opts)
+	if err != nil {
+		return nil, err
+	}
+
+	bus := &EventBus{
 		publisher: pub,
 		marshaler: marshaler,
 		namer:     namer,
 	}
+
+	if cfg.outbox != nil {
+		bus.forwarder = newForwarderState(cfg.outbox)
+		bus.publisher = bus.forwarder.wrapPublisher(pub)
+	}
+
+	return bus, nil
 }
 
 // validate checks that the event bus and its dependencies are properly initialized.
@@ -90,3 +118,21 @@ func (b *EventBus) Publish(ctx context.Context, evt any) error {
 
 	return b.publisher.Publish(topic, msg)
 }
+
+// RunForwarder starts the optional outbox forwarder when configured.
+func (b *EventBus) RunForwarder(ctx context.Context) error {
+	if b == nil || b.forwarder == nil {
+		return nil
+	}
+
+	return b.forwarder.Run(ctx)
+}
+
+// CloseForwarder stops the optional forwarder if it was started.
+func (b *EventBus) CloseForwarder(ctx context.Context) error {
+	if b == nil || b.forwarder == nil {
+		return nil
+	}
+
+	return b.forwarder.Close(ctx)
+}

cqrs/bus/options.go

@@ -0,0 +1,174 @@
+package bus
+
+import (
+	"database/sql"
+	"errors"
+	"fmt"
+	"strings"
+
+	wmmessage "github.com/ThreeDotsLabs/watermill/message"
+	"github.com/jackc/pgx/v5/pgxpool"
+	"github.com/jackc/pgx/v5/stdlib"
+	"github.com/shortlink-org/go-sdk/logger"
+	"go.opentelemetry.io/otel/metric"
+)
+
+const defaultForwarderTopic = "shortlink_cqrs_outbox"
+
+// Option configures Bus behaviours without breaking the constructor API.
+type Option func(*cqrsConfig)
+
+type cqrsConfig struct {
+	outbox *OutboxConfig
+	err    error
+}
+
+// OutboxConfig wires transactional outbox pieces required by the Watermill forwarder.
+type OutboxConfig struct {
+	DB            *sql.DB
+	Pool          *pgxpool.Pool
+	Subscriber    wmmessage.Subscriber
+	RealPublisher wmmessage.Publisher
+	ForwarderName string
+	Logger        logger.Logger
+	MeterProvider metric.MeterProvider
+}
+
+// WithOutbox enables Watermill's Outbox/Forwarder transport.
+//
+// Example with *sql.DB:
+//
+//	package main
+//
+//	import (
+//		"database/sql"
+//
+//		"github.com/ThreeDotsLabs/watermill-sql/v2/pkg/sql"
+//		"github.com/shortlink-org/go-sdk/cqrs/bus"
+//	)
+//
+//	func wireBuses(db *sql.DB, kafkaPub wmmessage.Publisher) *bus.CommandBus {
+//		sqlPublisher, _ := sql.NewPublisher(db, sqlPublisherConfig, sqlLogger)
+//		sqlSubscriber, _ := sql.NewSubscriber(db, sqlSubscriberConfig, sqlLogger)
+//
+//		return bus.NewCommandBus(
+//			sqlPublisher,
+//			marshaler,
+//			namer,
+//			bus.WithOutbox(bus.OutboxConfig{
+//				DB:            db,
+//				Subscriber:    sqlSubscriber,
+//				RealPublisher: kafkaPub,
+//				ForwarderName: "orders_outbox_forwarder",
+//				Logger:        log,
+//				MeterProvider: meterProvider,
+//			}),
+//		)
+//	}
+//
+// Example with *pgxpool.Pool (converted automatically):
+//
+//	package main
+//
+//	import (
+//		"database/sql"
+//
+//		"github.com/jackc/pgx/v5/pgxpool"
+//		"github.com/ThreeDotsLabs/watermill-sql/v2/pkg/sql"
+//		"github.com/shortlink-org/go-sdk/cqrs/bus"
+//	)
+//
+//	func wireBusesWithPgx(pool *pgxpool.Pool, kafkaPub wmmessage.Publisher) *bus.CommandBus {
+//		// Convert pgxpool to sql.DB for watermill-sql
+//		sqlDB, _ := sql.Open("pgx", pool.Config().ConnString())
+//		sqlPublisher, _ := sql.NewPublisher(sqlDB, sqlPublisherConfig, sqlLogger)
+//		sqlSubscriber, _ := sql.NewSubscriber(sqlDB, sqlSubscriberConfig, sqlLogger)
+//
+//		return bus.NewCommandBus(
+//			sqlPublisher,
+//			marshaler,
+//			namer,
+//			bus.WithOutbox(bus.OutboxConfig{
+//				Pool:          pool,
+//				Subscriber:    sqlSubscriber,
+//				RealPublisher: kafkaPub,
+//				ForwarderName: "orders_outbox_forwarder",
+//				Logger:        log,
+//				MeterProvider: meterProvider,
+//			}),
+//		)
+//	}
+func WithOutbox(cfg OutboxConfig) Option {
+	return func(c *cqrsConfig) {
+		conf := cfg
+		if c.err != nil {
+			return
+		}
+		if err := conf.prepare(); err != nil {
+			c.err = err
+			return
+		}
+		c.outbox = &conf
+	}
+}
+
+func applyOptions(opts []Option) (cqrsConfig, error) {
+	var cfg cqrsConfig
+	for _, opt := range opts {
+		if opt == nil {
+			continue
+		}
+		opt(&cfg)
+		if cfg.err != nil {
+			break
+		}
+	}
+	return cfg, cfg.err
+}
+
+func (c *OutboxConfig) prepare() error {
+	if c.DB == nil && c.Pool == nil {
+		return errors.New("cqrs/bus: sql.DB or pgxpool.Pool must be provided")
+	}
+	if c.Subscriber == nil {
+		return errors.New("cqrs/bus: outbox subscriber is required")
+	}
+	if c.RealPublisher == nil {
+		return errors.New("cqrs/bus: real publisher is required")
+	}
+	if c.Logger == nil {
+		return errors.New("cqrs/bus: logger is required")
+	}
+	if c.MeterProvider == nil {
+		return errors.New("cqrs/bus: meter provider is required")
+	}
+	if c.DB == nil && c.Pool != nil {
+		c.DB = stdlib.OpenDBFromPool(c.Pool)
+	}
+	c.ForwarderName = sanitizeForwarderTopic(c.ForwarderName, c.DB, c.Pool)
+	return nil
+}
+
+func sanitizeForwarderTopic(name string, db *sql.DB, pool *pgxpool.Pool) string {
+	name = strings.TrimSpace(name)
+	if name != "" {
+		return name
+	}
+
+	driverName := "sql"
+	if pool != nil {
+		driverName = "pgxpool"
+	} else if db != nil && db.Driver() != nil {
+		driverName = fmt.Sprintf("%T", db.Driver())
+	}
+
+	driverName = strings.ToLower(driverName)
+	driverName = strings.ReplaceAll(driverName, ".", "_")
+	driverName = strings.ReplaceAll(driverName, "*", "_")
+	driverName = strings.Trim(driverName, "_")
+	if driverName == "" {
+		driverName = "sql"
+	}
+
+	return fmt.Sprintf("%s_%s", defaultForwarderTopic, driverName)
+}