feat(cqrs): add JSON marshaler and examples

- Add JSONMarshaler for JSON payloads (alternative to ProtoMarshaler)
- Add comprehensive tests for JSONMarshaler
- Add basic example in cqrs/examples with JSON messages
- Remove unused gobreaker/v2 dependency
- Update registry to return errors on nil registration

Author: batazor

cqrs/bus/command_bus.go

@@ -32,8 +32,8 @@ func NewCommandBus(pub wmmessage.Publisher, marshaler cqrsmessage.Marshaler, nam
 	}
 }
 
-// Send encodes and publishes command with Shortlink metadata and tracing context.
-func (b *CommandBus) Send(ctx context.Context, cmd any) error {
+// validate checks that the command bus and its dependencies are properly initialized.
+func (b *CommandBus) validate(cmd any) error {
 	if b == nil {
 		return errCommandBusUninitialized
 	}
@@ -46,6 +46,14 @@ func (b *CommandBus) Send(ctx context.Context, cmd any) error {
 	if cmd == nil {
 		return errCommandPayloadNil
 	}
+	return nil
+}
+
+// Send encodes and publishes command with Shortlink metadata and tracing context.
+func (b *CommandBus) Send(ctx context.Context, cmd any) error {
+	if err := b.validate(cmd); err != nil {
+		return err
+	}
 	if ctx == nil {
 		ctx = context.Background()
 	}

cqrs/bus/event_bus.go

@@ -32,8 +32,8 @@ func NewEventBus(pub wmmessage.Publisher, marshaler cqrsmessage.Marshaler, namer
 	}
 }
 
-// Publish sends event using canonical topic name.
-func (b *EventBus) Publish(ctx context.Context, evt any) error {
+// validate checks that the event bus and its dependencies are properly initialized.
+func (b *EventBus) validate(evt any) error {
 	if b == nil {
 		return errEventBusUninitialized
 	}
@@ -46,6 +46,14 @@ func (b *EventBus) Publish(ctx context.Context, evt any) error {
 	if evt == nil {
 		return errEventPayloadNil
 	}
+	return nil
+}
+
+// Publish sends event using canonical topic name.
+func (b *EventBus) Publish(ctx context.Context, evt any) error {
+	if err := b.validate(evt); err != nil {
+		return err
+	}
 	if ctx == nil {
 		ctx = context.Background()
 	}

cqrs/bus/registry.go

@@ -30,11 +30,19 @@ func NewTypeRegistry() *TypeRegistry {
 	}
 }
 
-// RegisterCommand registers command type.
-func (r *TypeRegistry) RegisterCommand(cmd any) error {
+// validateCommand checks that command is not nil.
+func (r *TypeRegistry) validateCommand(cmd any) error {
 	if cmd == nil {
 		return ErrNilCommandType
 	}
+	return nil
+}
+
+// RegisterCommand registers command type.
+func (r *TypeRegistry) RegisterCommand(cmd any) error {
+	if err := r.validateCommand(cmd); err != nil {
+		return err
+	}
 	r.mu.Lock()
 	defer r.mu.Unlock()
 
@@ -43,11 +51,19 @@ func (r *TypeRegistry) RegisterCommand(cmd any) error {
 	return nil
 }
 
-// RegisterEvent registers event type.
-func (r *TypeRegistry) RegisterEvent(evt any) error {
+// validateEvent checks that event is not nil.
+func (r *TypeRegistry) validateEvent(evt any) error {
 	if evt == nil {
 		return ErrNilEventType
 	}
+	return nil
+}
+
+// RegisterEvent registers event type.
+func (r *TypeRegistry) RegisterEvent(evt any) error {
+	if err := r.validateEvent(evt); err != nil {
+		return err
+	}
 	r.mu.Lock()
 	defer r.mu.Unlock()
 

cqrs/bus/registry_test.go

@@ -5,6 +5,8 @@ import (
 	"sync"
 	"testing"
 
+	"github.com/stretchr/testify/require"
+
 	cqrsmessage "github.com/shortlink-org/go-sdk/cqrs/message"
 )
 
@@ -25,9 +27,8 @@ func TestTypeRegistryConcurrentAccess(t *testing.T) {
 					cqrsmessage.MetadataTypeVersion: "v1",
 				},
 			}
-			if err := reg.RegisterCommand(cmd); err != nil {
-				t.Errorf("register command %d: %v", i, err)
-			}
+			err := reg.RegisterCommand(cmd)
+			require.NoError(t, err, "register command %d", i)
 		}(i)
 
 		go func(i int) {
@@ -38,20 +39,18 @@ func TestTypeRegistryConcurrentAccess(t *testing.T) {
 					cqrsmessage.MetadataTypeVersion: "v1",
 				},
 			}
-			if err := reg.RegisterEvent(evt); err != nil {
-				t.Errorf("register event %d: %v", i, err)
-			}
+			err := reg.RegisterEvent(evt)
+			require.NoError(t, err, "register event %d", i)
 		}(i)
 	}
 
 	wg.Wait()
 
 	for i := 0; i < workers; i++ {
-		if _, ok := reg.ResolveCommand(fmt.Sprintf("billing.command.create_%d.v1", i)); !ok {
-			t.Fatalf("command %d not found", i)
-		}
-		if _, ok := reg.ResolveEvent(fmt.Sprintf("billing.aggregate.event_%d.v1", i)); !ok {
-			t.Fatalf("event %d not found", i)
-		}
+		_, ok := reg.ResolveCommand(fmt.Sprintf("billing.command.create_%d.v1", i))
+		require.True(t, ok, "command %d not found", i)
+
+		_, ok = reg.ResolveEvent(fmt.Sprintf("billing.aggregate.event_%d.v1", i))
+		require.True(t, ok, "event %d not found", i)
 	}
 }

cqrs/examples/README.md

@@ -0,0 +1,43 @@
+# CQRS Basic Example
+
+This example demonstrates the basic usage of the `cqrs` package with Watermill.
+
+## What it does
+
+- Registers command and event types in a type registry
+- Creates command and event buses
+- Sets up typed handlers for commands and events
+- Publishes a command and an event
+- Processes them through the router
+
+## Running the example
+
+```bash
+cd cqrs/examples
+go mod tidy
+go run main.go
+```
+
+## Expected output
+
+```
+🚀 Publishing CreateOrderCommand...
+📦 Processing command: CreateOrderCommand{OrderId: order-123, UserId: user-456, Amount: 99.99}
+✅ Order created successfully: order-123
+
+📢 Publishing OrderCreatedEvent...
+📊 Projecting event: OrderCreatedEvent{OrderId: order-123, UserId: user-456, Amount: 99.99}
+✅ Projection updated for order: order-123
+
+✨ Example completed successfully!
+```
+
+## Key concepts
+
+- **Type Registry**: Central registry for all commands and events
+- **Namer**: Generates canonical names like `orders.command.create_order.v1`
+- **Marshaler**: Serializes protobuf messages with metadata
+- **CommandBus/EventBus**: Publish commands and events with automatic metadata enrichment
+- **Handlers**: Typed handlers that receive unmarshaled protobuf messages
+- **Router**: Watermill router with CQRS middleware (retry, timeout, etc.)
+

cqrs/examples/go.mod

@@ -0,0 +1,28 @@
+module github.com/shortlink-org/go-sdk/cqrs/examples
+
+go 1.25.4
+
+replace github.com/shortlink-org/go-sdk/cqrs => ../
+
+replace github.com/shortlink-org/go-sdk/watermill => ../../watermill
+
+require (
+	github.com/ThreeDotsLabs/watermill v1.5.1
+	github.com/google/uuid v1.6.0
+	github.com/shortlink-org/go-sdk/cqrs v0.0.0-00010101000000-000000000000
+)
+
+require (
+	github.com/cenkalti/backoff/v5 v5.0.3 // indirect
+	github.com/go-logr/logr v1.4.3 // indirect
+	github.com/go-logr/stdr v1.2.2 // indirect
+	github.com/lithammer/shortuuid/v3 v3.0.7 // indirect
+	github.com/oklog/ulid v1.3.1 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/sony/gobreaker v1.0.0 // indirect
+	go.opentelemetry.io/auto/sdk v1.1.0 // indirect
+	go.opentelemetry.io/otel v1.38.0 // indirect
+	go.opentelemetry.io/otel/metric v1.38.0 // indirect
+	go.opentelemetry.io/otel/trace v1.38.0 // indirect
+	google.golang.org/protobuf v1.36.10 // indirect
+)

cqrs/examples/go.sum

@@ -0,0 +1,43 @@
+github.com/ThreeDotsLabs/watermill v1.5.1 h1:t5xMivyf9tpmU3iozPqyrCZXHvoV1XQDfihas4sV0fY=
+github.com/ThreeDotsLabs/watermill v1.5.1/go.mod h1:Uop10dA3VeJWsSvis9qO3vbVY892LARrKAdki6WtXS4=
+github.com/cenkalti/backoff/v5 v5.0.3 h1:ZN+IMa753KfX5hd8vVaMixjnqRZ3y8CuJKRKj1xcsSM=
+github.com/cenkalti/backoff/v5 v5.0.3/go.mod h1:rkhZdG3JZukswDf7f0cwqPNk4K0sa+F97BxZthm/crw=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/go-logr/logr v1.2.2/go.mod h1:jdQByPbusPIv2/zmleS9BjJVeZ6kBagPoEUsqbVz/1A=
+github.com/go-logr/logr v1.4.3 h1:CjnDlHq8ikf6E492q6eKboGOC0T8CDaOvkHCIg8idEI=
+github.com/go-logr/logr v1.4.3/go.mod h1:9T104GzyrTigFIr8wt5mBrctHMim0Nb2HLGrmQ40KvY=
+github.com/go-logr/stdr v1.2.2 h1:hSWxHoqTgW2S2qGc0LTAI563KZ5YKYRhT3MFKZMbjag=
+github.com/go-logr/stdr v1.2.2/go.mod h1:mMo/vtBO5dYbehREoey6XUKy/eSumjCCveDpRre4VKE=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/uuid v1.2.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/lithammer/shortuuid/v3 v3.0.7 h1:trX0KTHy4Pbwo/6ia8fscyHoGA+mf1jWbPJVuvyJQQ8=
+github.com/lithammer/shortuuid/v3 v3.0.7/go.mod h1:vMk8ke37EmiewwolSO1NLW8vP4ZaKlRuDIi8tWWmAts=
+github.com/oklog/ulid v1.3.1 h1:EGfNDEx6MqHz8B3uNV6QAib1UR2Lm97sHi3ocA6ESJ4=
+github.com/oklog/ulid v1.3.1/go.mod h1:CirwcVhetQ6Lv90oh/F+FBtV6XMibvdAFo93nm5qn4U=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/sony/gobreaker v1.0.0 h1:feX5fGGXSl3dYd4aHZItw+FpHLvvoaqkawKjVNiFMNQ=
+github.com/sony/gobreaker v1.0.0/go.mod h1:ZKptC7FHNvhBz7dN2LGjPVBz2sZJmc0/PkyDJOjmxWY=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+go.opentelemetry.io/auto/sdk v1.1.0 h1:cH53jehLUN6UFLY71z+NDOiNJqDdPRaXzTel0sJySYA=
+go.opentelemetry.io/auto/sdk v1.1.0/go.mod h1:3wSPjt5PWp2RhlCcmmOial7AvC4DQqZb7a7wCow3W8A=
+go.opentelemetry.io/otel v1.38.0 h1:RkfdswUDRimDg0m2Az18RKOsnI8UDzppJAtj01/Ymk8=
+go.opentelemetry.io/otel v1.38.0/go.mod h1:zcmtmQ1+YmQM9wrNsTGV/q/uyusom3P8RxwExxkZhjM=
+go.opentelemetry.io/otel/metric v1.38.0 h1:Kl6lzIYGAh5M159u9NgiRkmoMKjvbsKtYRwgfrA6WpA=
+go.opentelemetry.io/otel/metric v1.38.0/go.mod h1:kB5n/QoRM8YwmUahxvI3bO34eVtQf2i4utNVLr9gEmI=
+go.opentelemetry.io/otel/trace v1.38.0 h1:Fxk5bKrDZJUH+AMyyIXGcFAPah0oRcT+LuNtJrmcNLE=
+go.opentelemetry.io/otel/trace v1.38.0/go.mod h1:j1P9ivuFsTceSWe1oY+EeW3sc+Pp42sO++GHkg4wwhs=
+google.golang.org/protobuf v1.36.10 h1:AYd7cD/uASjIL6Q9LiTjz8JLcrh/88q5UObnmY3aOOE=
+google.golang.org/protobuf v1.36.10/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=

cqrs/examples/handlers.go

@@ -0,0 +1,36 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"time"
+)
+
+// CreateOrderHandler processes CreateOrderCommand
+type CreateOrderHandler struct{}
+
+func (h *CreateOrderHandler) Handle(ctx context.Context, cmd *CreateOrderCommand) error {
+	fmt.Printf("📦 Processing command: CreateOrderCommand{OrderId: %s, UserId: %s, Amount: %.2f}\n",
+		cmd.OrderId, cmd.UserId, cmd.Amount)
+
+	// Simulate business logic
+	time.Sleep(100 * time.Millisecond)
+
+	// In real app, you would publish events here
+	fmt.Printf("✅ Order created successfully: %s\n", cmd.OrderId)
+	return nil
+}
+
+// OrderCreatedProjector processes OrderCreatedEvent
+type OrderCreatedProjector struct{}
+
+func (p *OrderCreatedProjector) Handle(ctx context.Context, evt *OrderCreatedEvent) error {
+	fmt.Printf("📊 Projecting event: OrderCreatedEvent{OrderId: %s, UserId: %s, Amount: %.2f}\n",
+		evt.OrderId, evt.UserId, evt.Amount)
+
+	// Simulate projection logic (e.g., update read model)
+	time.Sleep(50 * time.Millisecond)
+
+	fmt.Printf("✅ Projection updated for order: %s\n", evt.OrderId)
+	return nil
+}