Microservices Patterns

package main

import (
	"fmt"
	"math"
	"math/rand"
	"sync"
	"time"
)

// --- Types ---
type SagaStatus string
type StepStatus string

const (
	StatusPending      SagaStatus = "pending"
	StatusRunning      SagaStatus = "running"
	StatusCompleted    SagaStatus = "completed"
	StatusCompensating SagaStatus = "compensating"
	StatusFailed       SagaStatus = "failed"

	StepPending     StepStatus = "pending"
	StepSuccess     StepStatus = "success"
	StepFailed      StepStatus = "failed"
	StepCompensated StepStatus = "compensated"
)

type SagaStep struct {
	Name       string
	Execute    func(ctx map[string]interface{}) (map[string]interface{}, error)
	Compensate func(ctx map[string]interface{}) error
}

type StepState struct {
	Name   string     `json:"name"`
	Status StepStatus `json:"status"`
	Error  string     `json:"error,omitempty"`
}

type SagaState struct {
	ID          string                 `json:"id"`
	Status      SagaStatus             `json:"status"`
	Steps       []StepState            `json:"steps"`
	Context     map[string]interface{} `json:"context"`
	StartedAt   time.Time              `json:"startedAt"`
	CompletedAt *time.Time             `json:"completedAt,omitempty"`
}

// --- Circuit Breaker ---
type CircuitState int

const (
	CircuitClosed CircuitState = iota
	CircuitOpen
	CircuitHalfOpen
)

func (cs CircuitState) String() string {
	switch cs {
	case CircuitClosed:
		return "CLOSED"
	case CircuitOpen:
		return "OPEN"
	case CircuitHalfOpen:
		return "HALF_OPEN"
	default:
		return "UNKNOWN"
	}
}

type CircuitBreaker struct {
	mu               sync.Mutex
	name             string
	state            CircuitState
	failureCount     int
	successCount     int
	failureThreshold int
	resetTimeout     time.Duration
	lastFailureTime  time.Time
	halfOpenMax      int
}

func NewCircuitBreaker(name string, threshold int, resetTimeout time.Duration) *CircuitBreaker {
	return &CircuitBreaker{
		name:             name,
		state:            CircuitClosed,
		failureThreshold: threshold,
		resetTimeout:     resetTimeout,
		halfOpenMax:      3,
	}
}

func (cb *CircuitBreaker) Call(fn func() (map[string]interface{}, error)) (map[string]interface{}, error) {
	cb.mu.Lock()

	if cb.state == CircuitOpen {
		if time.Since(cb.lastFailureTime) > cb.resetTimeout {
			cb.state = CircuitHalfOpen
			cb.successCount = 0
			fmt.Printf("[CIRCUIT:%s] OPEN -> HALF_OPEN\n", cb.name)
		} else {
			cb.mu.Unlock()
			return nil, fmt.Errorf("circuit breaker %s is OPEN", cb.name)
		}
	}
	currentState := cb.state
	cb.mu.Unlock()

	result, err := fn()

	cb.mu.Lock()
	defer cb.mu.Unlock()

	if err != nil {
		cb.failureCount++
		cb.lastFailureTime = time.Now()
		if cb.failureCount >= cb.failureThreshold {
			cb.state = CircuitOpen
			fmt.Printf("[CIRCUIT:%s] -> OPEN (failures: %d)\n", cb.name, cb.failureCount)
		}
		return nil, err
	}

	if currentState == CircuitHalfOpen {
		cb.successCount++
		if cb.successCount >= cb.halfOpenMax {
			cb.state = CircuitClosed
			cb.failureCount = 0
			fmt.Printf("[CIRCUIT:%s] HALF_OPEN -> CLOSED\n", cb.name)
		}
	} else {
		cb.failureCount = 0
	}

	return result, nil
}

// --- Service Registry ---
type ServiceInstance struct {
	ID              string `json:"id"`
	Name            string `json:"name"`
	URL             string `json:"url"`
	Healthy         bool   `json:"healthy"`
	LastHealthCheck time.Time
}

type ServiceRegistry struct {
	mu       sync.RWMutex
	services map[string][]ServiceInstance
	stopCh   chan struct{}
}

func NewServiceRegistry() *ServiceRegistry {
	return &ServiceRegistry{
		services: make(map[string][]ServiceInstance),
		stopCh:   make(chan struct{}),
	}
}

func (sr *ServiceRegistry) Register(name, url string) string {
	sr.mu.Lock()
	defer sr.mu.Unlock()

	id := fmt.Sprintf("%s-%d", name, time.Now().UnixNano()%100000)
	instance := ServiceInstance{
		ID:              id,
		Name:            name,
		URL:             url,
		Healthy:         true,
		LastHealthCheck: time.Now(),
	}
	sr.services[name] = append(sr.services[name], instance)
	fmt.Printf("[REGISTRY] Registered %s at %s (id: %s)\n", name, url, id)
	return id
}

func (sr *ServiceRegistry) Resolve(name string) (*ServiceInstance, error) {
	sr.mu.RLock()
	defer sr.mu.RUnlock()

	instances := sr.services[name]
	if len(instances) == 0 {
		return nil, fmt.Errorf("no instances for service %s", name)
	}

	var healthy []ServiceInstance
	for _, inst := range instances {
		if inst.Healthy {
			healthy = append(healthy, inst)
		}
	}
	if len(healthy) == 0 {
		return nil, fmt.Errorf("no healthy instances for service %s", name)
	}

	picked := healthy[rand.Intn(len(healthy))]
	return &picked, nil
}

func (sr *ServiceRegistry) StartHealthChecks(interval time.Duration) {
	go func() {
		ticker := time.NewTicker(interval)
		defer ticker.Stop()
		for {
			select {
			case <-ticker.C:
				sr.mu.RLock()
				for _, instances := range sr.services {
					for _, inst := range instances {
						status := "UP"
						if !inst.Healthy {
							status = "DOWN"
						}
						fmt.Printf("[HEALTH] Checking %s: %s\n", inst.ID, status)
					}
				}
				sr.mu.RUnlock()
			case <-sr.stopCh:
				return
			}
		}
	}()
}

func (sr *ServiceRegistry) Stop() {
	close(sr.stopCh)
}

// --- Retry with exponential backoff ---
func retryWithBackoff(fn func() (map[string]interface{}, error), maxRetries int, baseDelay time.Duration) (map[string]interface{}, error) {
	var lastErr error

	for attempt := 0; attempt <= maxRetries; attempt++ {
		result, err := fn()
		if err == nil {
			return result, nil
		}
		lastErr = err

		if attempt == maxRetries {
			break
		}

		delay := time.Duration(float64(baseDelay) * math.Pow(2, float64(attempt)))
		jitter := time.Duration(rand.Float64() * float64(time.Second))
		sleepTime := delay + jitter

		fmt.Printf("[RETRY] Attempt %d/%d failed, retrying in %v\n", attempt+1, maxRetries, sleepTime)
		time.Sleep(sleepTime)
	}

	return nil, lastErr
}

// --- Saga Orchestrator ---
type SagaOrchestrator struct {
	mu       sync.RWMutex
	sagas    map[string]*SagaState
	breakers map[string]*CircuitBreaker
	registry *ServiceRegistry
}

func NewSagaOrchestrator(registry *ServiceRegistry) *SagaOrchestrator {
	return &SagaOrchestrator{
		sagas:    make(map[string]*SagaState),
		breakers: make(map[string]*CircuitBreaker),
		registry: registry,
	}
}

func (so *SagaOrchestrator) getBreaker(name string) *CircuitBreaker {
	so.mu.Lock()
	defer so.mu.Unlock()
	if cb, ok := so.breakers[name]; ok {
		return cb
	}
	cb := NewCircuitBreaker(name, 5, 30*time.Second)
	so.breakers[name] = cb
	return cb
}

func (so *SagaOrchestrator) Execute(sagaID string, steps []SagaStep, initialCtx map[string]interface{}) *SagaState {
	state := &SagaState{
		ID:        sagaID,
		Status:    StatusRunning,
		Steps:     make([]StepState, len(steps)),
		Context:   copyMap(initialCtx),
		StartedAt: time.Now(),
	}
	for i, step := range steps {
		state.Steps[i] = StepState{Name: step.Name, Status: StepPending}
	}

	so.mu.Lock()
	so.sagas[sagaID] = state
	so.mu.Unlock()

	fmt.Printf("\n[SAGA:%s] Starting saga with %d steps\n", sagaID, len(steps))

	completedSteps := 0

	for i, step := range steps {
		breaker := so.getBreaker(step.Name)
		fmt.Printf("[SAGA:%s] Step %d/%d: %s -> EXECUTING\n", sagaID, i+1, len(steps), step.Name)

		result, err := retryWithBackoff(func() (map[string]interface{}, error) {
			return breaker.Call(func() (map[string]interface{}, error) {
				return step.Execute(state.Context)
			})
		}, 3, time.Second)

		if err != nil {
			state.Steps[i].Status = StepFailed
			state.Steps[i].Error = err.Error()
			fmt.Printf("[SAGA:%s] Step %s -> FAILED: %v\n", sagaID, step.Name, err)

			// Compensate
			state.Status = StatusCompensating
			fmt.Printf("[SAGA:%s] Starting compensation for %d completed steps\n", sagaID, completedSteps)

			for j := completedSteps - 1; j >= 0; j-- {
				compStep := steps[j]
				fmt.Printf("[SAGA:%s] Compensating: %s\n", sagaID, compStep.Name)
				if compErr := compStep.Compensate(state.Context); compErr != nil {
					fmt.Printf("[SAGA:%s] Compensation FAILED for %s: %v\n", sagaID, compStep.Name, compErr)
				} else {
					state.Steps[j].Status = StepCompensated
					fmt.Printf("[SAGA:%s] %s -> COMPENSATED\n", sagaID, compStep.Name)
				}
			}

			state.Status = StatusFailed
			now := time.Now()
			state.CompletedAt = &now
			fmt.Printf("[SAGA:%s] Saga FAILED (took %v)\n", sagaID, now.Sub(state.StartedAt))
			return state
		}

		// Merge result into context
		for k, v := range result {
			state.Context[k] = v
		}
		state.Steps[i].Status = StepSuccess
		completedSteps = i + 1
		fmt.Printf("[SAGA:%s] Step %s -> SUCCESS\n", sagaID, step.Name)
	}

	state.Status = StatusCompleted
	now := time.Now()
	state.CompletedAt = &now
	fmt.Printf("[SAGA:%s] Saga COMPLETED (took %v)\n\n", sagaID, now.Sub(state.StartedAt))
	return state
}

func copyMap(src map[string]interface{}) map[string]interface{} {
	dst := make(map[string]interface{}, len(src))
	for k, v := range src {
		dst[k] = v
	}
	return dst
}

// --- Define Order Saga steps ---
func createOrderSagaSteps() []SagaStep {
	return []SagaStep{
		{
			Name: "CreateOrder",
			Execute: func(ctx map[string]interface{}) (map[string]interface{}, error) {
				fmt.Printf("  Creating order for user %v\n", ctx["userId"])
				orderID := fmt.Sprintf("ORD-%d", time.Now().UnixMilli())
				return map[string]interface{}{"orderId": orderID, "orderStatus": "created"}, nil
			},
			Compensate: func(ctx map[string]interface{}) error {
				fmt.Printf("  Cancelling order %v\n", ctx["orderId"])
				return nil
			},
		},
		{
			Name: "ReserveInventory",
			Execute: func(ctx map[string]interface{}) (map[string]interface{}, error) {
				fmt.Printf("  Reserving inventory for order %v\n", ctx["orderId"])
				resID := fmt.Sprintf("RES-%d", time.Now().UnixMilli())
				return map[string]interface{}{"inventoryReserved": true, "reservationId": resID}, nil
			},
			Compensate: func(ctx map[string]interface{}) error {
				fmt.Printf("  Releasing inventory reservation %v\n", ctx["reservationId"])
				return nil
			},
		},
		{
			Name: "ChargePayment",
			Execute: func(ctx map[string]interface{}) (map[string]interface{}, error) {
				amount, _ := ctx["amount"].(float64)
				fmt.Printf("  Charging payment for order %v, amount: $%.2f\n", ctx["orderId"], amount)
				if amount > 1000 {
					return nil, fmt.Errorf("payment declined: insufficient funds")
				}
				payID := fmt.Sprintf("PAY-%d", time.Now().UnixMilli())
				return map[string]interface{}{"paymentId": payID, "charged": true}, nil
			},
			Compensate: func(ctx map[string]interface{}) error {
				fmt.Printf("  Refunding payment %v\n", ctx["paymentId"])
				return nil
			},
		},
		{
			Name: "ConfirmOrder",
			Execute: func(ctx map[string]interface{}) (map[string]interface{}, error) {
				fmt.Printf("  Confirming order %v (payment: %v)\n", ctx["orderId"], ctx["paymentId"])
				return map[string]interface{}{
					"orderStatus": "confirmed",
					"confirmedAt": time.Now().Format(time.RFC3339),
				}, nil
			},
			Compensate: func(ctx map[string]interface{}) error {
				fmt.Printf("  Reverting order %v confirmation\n", ctx["orderId"])
				return nil
			},
		},
	}
}

// --- Main ---
func main() {
	registry := NewServiceRegistry()
	registry.Register("order-service", "http://localhost:3001")
	registry.Register("inventory-service", "http://localhost:3002")
	registry.Register("payment-service", "http://localhost:3003")

	orchestrator := NewSagaOrchestrator(registry)
	steps := createOrderSagaSteps()

	// Successful order
	fmt.Println("=== Scenario 1: Successful Order ===")
	orchestrator.Execute("saga-001", steps, map[string]interface{}{
		"userId": "user-42",
		"items":  []map[string]interface{}{{"sku": "WIDGET-A", "qty": 2}, {"sku": "GADGET-B", "qty": 1}},
		"amount": 99.99,
	})

	// Failed order -- triggers compensation
	fmt.Println("=== Scenario 2: Payment Failure with Rollback ===")
	orchestrator.Execute("saga-002", steps, map[string]interface{}{
		"userId": "user-42",
		"items":  []map[string]interface{}{{"sku": "EXPENSIVE-ITEM", "qty": 1}},
		"amount": 5000.0,
	})

	registry.Stop()
}