Composable Commerce Architecture: Beyond Tech Debt

static_resources:
  listeners:
  - name: ingress_listener
    address:
      socket_address: { address: 0.0.0.0, port_value: 8080 }
    filter_chains:
    - filters:
      - name: envoy.filters.network.http_connection_manager
        typed_config:
          "@type": type.googleapis.com/envoy.extensions.filters.network.http_connection_manager.v3.HttpConnectionManager
          stat_prefix: ingress_http
          route_config:
            name: dynamic_ecommerce_routes
            virtual_hosts:
            - name: api_upstream
              domains: ["api.tanhdev.com", "store.tanhdev.com"]
              routes:
              # 1. Fully migrated Bounded Context (Cart & Inventory)
              - match: { prefix: "/api/v1/cart" }
                route:
                  cluster: composable_bff_cluster
                  timeout: 3s
                  retry_policy:
                    retry_on: "5xx,connect-failure,refused-stream"
                    num_retries: 3
                    per_try_timeout: 800ms

              # 2. Bounded Context under canary trial & shadow mirroring (Order Service)
              - match: { prefix: "/api/v1/orders" }
                route:
                  weighted_clusters:
                    clusters:
                    - name: legacy_monolith_cluster
                      weight: 80
                    - name: composable_bff_cluster
                      weight: 20
                  # Mirror 10% of traffic quietly to the new cluster for benchmarking
                  request_mirror_policies:
                  - cluster: composable_order_shadow_cluster
                    runtime_fraction:
                      default_value: { numerator: 10, denominator: HUNDRED }

              # 3. Default Fallback: all remaining routes point to the Monolith
              - match: { prefix: "/" }
                route: { cluster: legacy_monolith_cluster, timeout: 15s }
          http_filters:
          - name: envoy.filters.http.router
            typed_config:
              "@type": type.googleapis.com/envoy.extensions.filters.http.router.v3.Router

  clusters:
  - name: legacy_monolith_cluster
    connect_timeout: 0.50s
    type: LOGICAL_DNS
    dns_lookup_family: V4_ONLY
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: legacy_monolith_cluster
      endpoints:
      - lb_endpoints:
        - endpoint: { address: { socket_address: { address: monolith-prod.internal.net, port_value: 80 } } }

  - name: composable_bff_cluster
    connect_timeout: 0.25s
    type: LOGICAL_DNS
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: composable_bff_cluster
      endpoints:
      - lb_endpoints:
        - endpoint: { address: { socket_address: { address: bff-prod.internal.net, port_value: 443 } } }

  - name: composable_order_shadow_cluster
    connect_timeout: 0.25s
    type: LOGICAL_DNS
    lb_policy: ROUND_ROBIN
    load_assignment:
      cluster_name: composable_order_shadow_cluster
      endpoints:
      - lb_endpoints:
        - endpoint: { address: { socket_address: { address: order-shadow.internal.net, port_value: 443 } } }

package cartlock

import (
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"time"

	"github.com/redis/go-redis/v9"
)

var (
	ErrLockAcquisitionFailed = errors.New("cartlock: cannot acquire redis lock, cart is busy")
	ErrLockContextCancelled  = errors.New("cartlock: context cancelled before the lock could be acquired")

	// Lua Script guarantees the lock release is atomic:
	// Only delete the key if the value stored in Redis exactly matches the current process's token.
	luaReleaseScript = redis.NewScript(`
		if redis.call("get", KEYS[1]) == ARGV[1] then
			return redis.call("del", KEYS[1])
		else
			return 0
		end
	`)
)

type CartLocker struct {
	client *redis.Client
}

func NewCartLocker(client *redis.Client) *CartLocker {
	return &CartLocker{client: client}
}

func generateSecureToken() (string, error) {
	b := make([]byte, 16)
	if _, err := rand.Read(b); err != nil {
		return "", err
	}
	return hex.EncodeToString(b), nil
}

// AcquireLock acquires the Redis lock using the SET NX PX command
func (l *CartLocker) AcquireLock(ctx context.Context, cartID string, ttl time.Duration) (string, error) {
	select {
	case <-ctx.Done():
		return "", ErrLockContextCancelled
	default:
	}

	token, err := generateSecureToken()
	if err != nil {
		return "", fmt.Errorf("failed to generate a random token: %w", err)
	}

	lockKey := fmt.Sprintf("lock:cart:%s", cartID)
	ok, err := l.client.SetNX(ctx, lockKey, token, ttl).Result()
	if err != nil {
		return "", fmt.Errorf("redis setnx error: %w", err)
	}
	if !ok {
		return "", ErrLockAcquisitionFailed
	}

	return token, nil
}

// ReleaseLock releases the lock atomically via the Lua script
func (l *CartLocker) ReleaseLock(ctx context.Context, cartID string, token string) (bool, error) {
	lockKey := fmt.Sprintf("lock:cart:%s", cartID)
	res, err := luaReleaseScript.Run(ctx, l.client, []string{lockKey}, token).Int64()
	if err != nil {
		return false, fmt.Errorf("lua release script execution error: %w", err)
	}
	return res == 1, nil
}

// WithCartLock wraps the full lifecycle of acquiring the lock, running the callback, and releasing the lock automatically
func (l *CartLocker) WithCartLock(
	ctx context.Context,
	cartID string,
	ttl time.Duration,
	action func(ctx context.Context) error,
) error {
	token, err := l.AcquireLock(ctx, cartID, ttl)
	if err != nil {
		return err
	}

	defer func() {
		// Use an independent context with a short timeout to guarantee the lock is always cleaned up
		// even when the action itself times out
		cleanupCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		_, _ = l.ReleaseLock(cleanupCtx, cartID, token)
	}()

	return action(ctx)
}

package main

import (
	"context"
	"database/sql"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/IBM/sarama"
	"github.com/lib/pq"
)

type OutboxEvent struct {
	ID        string    `json:"id"`
	Aggregate string    `json:"aggregate_type"`
	Payload   string    `json:"payload"`
	CreatedAt time.Time `json:"created_at"`
}

type OutboxPoller struct {
	db       *sql.DB
	producer sarama.SyncProducer
	topic    string
}

func NewOutboxPoller(db *sql.DB, producer sarama.SyncProducer, topic string) *OutboxPoller {
	return &OutboxPoller{db: db, producer: producer, topic: topic}
}

// ProcessBatch processes pending events in batches using the high-performance SKIP LOCKED syntax
func (p *OutboxPoller) ProcessBatch(ctx context.Context, batchSize int) (int, error) {
	tx, err := p.db.BeginTx(ctx, &sql.TxOptions{Isolation: sql.LevelReadCommitted})
	if err != nil {
		return 0, fmt.Errorf("failed to begin transaction: %w", err)
	}
	defer tx.Rollback()

	// Lock unprocessed rows and skip rows currently locked by other workers
	query := `
		SELECT id, aggregate_type, payload, created_at
		FROM outbox_events
		WHERE processed_at IS NULL
		ORDER BY created_at ASC
		LIMIT $1
		FOR UPDATE SKIP LOCKED
	`
	rows, err := tx.QueryContext(ctx, query, batchSize)
	if err != nil {
		return 0, fmt.Errorf("failed to query outbox events: %w", err)
	}
	defer rows.Close()

	var events []OutboxEvent
	for rows.Next() {
		var ev OutboxEvent
		if err := rows.Scan(&ev.ID, &ev.Aggregate, &ev.Payload, &ev.CreatedAt); err != nil {
			return 0, fmt.Errorf("failed to scan outbox row: %w", err)
		}
		events = append(events, ev)
	}

	if len(events) == 0 {
		return 0, nil
	}

	// Publish to Kafka in bulk
	for _, ev := range events {
		msg := &sarama.ProducerMessage{
			Topic: p.topic,
			Key:   sarama.StringEncoder(ev.Aggregate),
			Value: sarama.StringEncoder(ev.Payload),
			Headers: []sarama.RecordHeader{
				{Key: []byte("outbox_id"), Value: []byte(ev.ID)},
				{Key: []byte("produced_at"), Value: []byte(time.Now().UTC().Format(time.RFC3339))},
			},
		}

		partition, offset, err := p.producer.SendMessage(msg)
		if err != nil {
			return 0, fmt.Errorf("failed to publish message %s to kafka: %w", ev.ID, err)
		}
		log.Printf("Published Outbox Event %s to partition %d, offset %d", ev.ID, partition, offset)
	}

	// Mark the successfully published events
	updateQuery := `
		UPDATE outbox_events
		SET processed_at = NOW()
		WHERE id = ANY($1)
	`
	ids := make([]string, len(events))
	for i, ev := range events {
		ids[i] = ev.ID
	}

	if _, err := tx.ExecContext(ctx, updateQuery, pq.Array(ids)); err != nil {
		return 0, fmt.Errorf("failed to update processed_at: %w", err)
	}

	if err := tx.Commit(); err != nil {
		return 0, fmt.Errorf("failed to commit the outbox transaction: %w", err)
	}

	return len(events), nil
}

func (p *OutboxPoller) Run(ctx context.Context, interval time.Duration, batchSize int) {
	ticker := time.NewTicker(interval)
	defer ticker.Stop()

	for {
		select {
		case <-ctx.Done():
			log.Println("Outbox Poller shutting down gracefully...")
			return
		case <-ticker.C:
			count, err := p.ProcessBatch(ctx, batchSize)
			if err != nil {
				log.Printf("Outbox batch processing error: %v", err)
			} else if count > 0 {
				log.Printf("Successfully processed %d outbox events", count)
			}
		}
	}
}

{
  "name": "monolith-inventory-cdc-connector",
  "config": {
    "connector.class": "io.debezium.connector.mysql.MySqlConnector",
    "tasks.max": "1",
    "database.hostname": "mysql-monolith-primary.internal.net",
    "database.port": "3306",
    "database.user": "debezium_cdc_user",
    "database.password": "${file:/secrets/db-credentials.properties:cdc_password}",
    "database.server.id": "184055",
    "database.include.list": "magento_production",
    "table.include.list": "magento_production.cataloginventory_stock_status,magento_production.sales_order",
    "topic.prefix": "cdc_legacy",
    "schema.history.internal.kafka.bootstrap.servers": "kafka-cluster.internal:9092",
    "schema.history.internal.kafka.topic": "schema-changes.inventory",
    "snapshot.mode": "when_needed",
    "snapshot.locking.mode": "minimal",
    "gtid.source.includes": "3E11FA47-71CA-11E1-9E33-C80AA9429A4E:1-100",
    "decimal.handling.mode": "double",
    "tombstones.on.delete": "true",
    "inconsistent.schema.handling.mode": "warn",
    "transforms": "unwrap,reroute",
    "transforms.unwrap.type": "io.debezium.transforms.ExtractNewRecordState",
    "transforms.unwrap.drop.tombstones": "false",
    "transforms.unwrap.delete.handling.mode": "rewrite",
    "transforms.reroute.type": "io.debezium.transforms.ByLogicalTableRouter",
    "transforms.reroute.topic.regex": "cdc_legacy\\.magento_production\\.(.*)",
    "transforms.reroute.topic.replacement": "events-cdc-$1-v1"
  }
}

package telemetry

import (
	"context"
	"fmt"
	"time"

	"go.opentelemetry.io/otel"
	"go.opentelemetry.io/otel/exporters/otlp/otlptrace/otlptracegrpc"
	"go.opentelemetry.io/otel/propagation"
	"go.opentelemetry.io/otel/sdk/resource"
	sdktrace "go.opentelemetry.io/otel/sdk/trace"
	semconv "go.opentelemetry.io/otel/semconv/v1.24.0"
	"google.golang.org/grpc"
	"google.golang.org/grpc/credentials/insecure"
)

// InitTracer bootstraps the OpenTelemetry TracerProvider connected to the OTel Collector over gRPC
func InitTracer(ctx context.Context, serviceName string, collectorAddr string) (func(context.Context) error, error) {
	res, err := resource.New(ctx,
		resource.WithAttributes(
			semconv.ServiceNameKey.String(serviceName),
			semconv.ServiceVersionKey.String("2.4.0"),
			semconv.DeploymentEnvironmentKey.String("production"),
		),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to initialize the otel resource: %w", err)
	}

	conn, err := grpc.DialContext(ctx, collectorAddr,
		grpc.WithTransportCredentials(insecure.NewCredentials()),
		grpc.WithBlock(),
		grpc.WithTimeout(3*time.Second),
	)
	if err != nil {
		return nil, fmt.Errorf("cannot connect to the otel collector over gRPC at %s: %w", collectorAddr, err)
	}

	exporter, err := otlptracegrpc.New(ctx, otlptracegrpc.WithGRPCConn(conn))
	if err != nil {
		return nil, fmt.Errorf("failed to initialize the otlp trace exporter: %w", err)
	}

	bsp := sdktrace.NewBatchSpanProcessor(exporter,
		sdktrace.WithBatchTimeout(5*time.Second),
		sdktrace.WithMaxExportBatchSize(512),
	)

	tp := sdktrace.NewTracerProvider(
		sdktrace.WithSampler(sdktrace.AlwaysSample()),
		sdktrace.WithResource(res),
		sdktrace.WithSpanProcessor(bsp),
	)

	// Register the TracerProvider and the standard W3C TextMapPropagator globally
	otel.SetTracerProvider(tp)
	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
		propagation.TraceContext{},
		propagation.Baggage{},
	))

	return tp.Shutdown, nil
}