Kiến trúc Ecommerce 2026: Vượt Qua Nợ Kỹ Thuật Khi Chuyển Đổi

package cartlock

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

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

var (
	ErrLockAcquisitionFailed = errors.New("cartlock: unable to acquire redis lock, cart state is currently busy")
	
	// Atomic Lua Release Script: Checks if key value matches token before deleting to prevent unlocking another instance's lock
	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}
}

// generateToken creates a 16-byte cryptographically secure random string
func generateToken() (string, error) {
	b := make([]byte, 16)
	_, err := rand.Read(b)
	if err != nil {
		return "", err
	}
	return hex.EncodeToString(b), nil
}

// AcquireLock attempts to set a distributed lock key in Redis using `SET NX PX`
func (l *CartLocker) AcquireLock(ctx context.Context, cartID string, ttl time.Duration) (string, error) {
	token, err := generateToken()
	if err != nil {
		return "", fmt.Errorf("failed to generate lock token: %w", err)
	}

	lockKey := fmt.Sprintf("lock:cart:%s", cartID)
	
	// SET lock:cart:<cartID> <token> NX PX <ttl_ms>
	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 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("redis lua release lock error: %w", err)
	}

	return res == 1, nil
}

// WithCartLock wraps a mutation callback function with automatic acquisition and atomic lock release
func (l *CartLocker) WithCartLock(ctx context.Context, cartID string, ttl time.Duration, fn func(ctx context.Context) error) error {
	token, err := l.AcquireLock(ctx, cartID, ttl)
	if err != nil {
		return err
	}

	defer func() {
		// Use a detached background context for lock release to guarantee cleanup even if parent ctx is cancelled
		releaseCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second)
		defer cancel()
		_, _ = l.ReleaseLock(releaseCtx, cartID, token)
	}()

	return fn(ctx)
}

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 initializes OpenTelemetry SDK with OTLP gRPC Exporter and global W3C context propagators
func InitTracer(ctx context.Context, serviceName, serviceVersion, otlpEndpoint string) (func(context.Context) error, error) {
	// 1. Create OTLP gRPC Exporter targeting Tempo / Jaeger / OpenTelemetry Collector
	exporter, err := otlptracegrpc.New(ctx,
		otlptracegrpc.WithEndpoint(otlpEndpoint),
		otlptracegrpc.WithTLSCredentials(insecure.NewCredentials()), // Use WithTLSCredentials for TLS in prod
		otlptracegrpc.WithDialOption(grpc.WithBlock()),
		otlptracegrpc.WithTimeout(5*time.Second),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create OTLP gRPC trace exporter: %w", err)
	}

	// 2. Define Service Metadata Resource
	res, err := resource.New(ctx,
		resource.WithAttributes(
			semconv.ServiceNameKey.String(serviceName),
			semconv.ServiceVersionKey.String(serviceVersion),
			semconv.DeploymentEnvironmentKey.String("production"),
		),
	)
	if err != nil {
		return nil, fmt.Errorf("failed to create otel resource: %w", err)
	}

	// 3. Construct TracerProvider with BatchSpanProcessor
	bsp := sdktrace.NewBatchSpanProcessor(exporter)
	tp := sdktrace.NewTracerProvider(
		sdktrace.WithSampler(sdktrace.AlwaysSample()), // Use ParentBased(TraceIDRatioBased(0.1)) in ultra-high traffic
		sdktrace.WithResource(res),
		sdktrace.WithSpanProcessor(bsp),
	)

	// 4. Register Global OpenTelemetry TracerProvider & W3C TraceContext Propagator
	otel.SetTracerProvider(tp)
	otel.SetTextMapPropagator(propagation.NewCompositeTextMapPropagator(
		propagation.TraceContext{},
		propagation.Baggage{},
	))

	// Return graceful shutdown closure
	return tp.Shutdown, nil
}

{
  "name": "legacy-mysql-inventory-cdc",
  "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",
    "database.password": "${file:/secrets/db-credentials.properties:cdc_password}",
    "database.server.id": "184054",
    "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"
  }
}