// Sistema de Worker Pool con Pipeline Pattern
// Implementación de un sistema de procesamiento de trabajos que combina Worker Pool y Pipeline para procesar datos de manera eficiente y escalable
package main
import (
	"context"
	"errors"
	"fmt"
	"sync"
	"sync/atomic"
	"time"
)
// Job representa un trabajo a procesar, en este contexto, una imagen a procesar.
type Job struct {
	ID   int
	Data string // Simula datos de imagen
}
// Stage es una función que procesa un job y produce output o error.
type Stage func(ctx context.Context, job Job) (Job, error)
// Worker es un trabajador en el pool.
type Worker struct {
	ID int
}
// WorkerPool gestiona un pool de workers.
type WorkerPool struct {
	size    int
	ctx     context.Context
	cancel  context.CancelFunc
	wg      sync.WaitGroup
	metrics struct {
		throughput atomic.Int64
		latency    atomic.Int64
	}
}
// NewWorkerPool crea un pool con número específico de workers.
func NewWorkerPool(size int, parentCtx context.Context) *WorkerPool {
	ctx, cancel := context.WithCancel(parentCtx)
	return &WorkerPool{
		size:   size,
		ctx:    ctx,
		cancel: cancel,
	}
}
// Process ejecuta el trabajo en un worker con manejo de errores.
func (w *Worker) Process(ctx context.Context, job Job, stage Stage) (Job, error) {
	select {
	case <-ctx.Done():
		return Job{}, ctx.Err()
	default:
		return stage(ctx, job)
	}
}
// Pipeline gestiona stages conectados por channels.
type Pipeline struct {
	stages []Stage
	input  <-chan Job
	output chan Job
	errs   chan error
	pool   *WorkerPool
	wg     sync.WaitGroup
}
// ProcessPipeline configura el pipeline de procesamiento.
func ProcessPipeline(input <-chan Job, stages []Stage, pool *WorkerPool) *Pipeline {
	return &Pipeline{
		stages: stages,
		input:  input,
		output: make(chan Job, len(stages)), // Buffered channel
		errs:   make(chan error, len(stages)),
		pool:   pool,
	}
}
// Start inicia todos los stages del pipeline.
func (p *Pipeline) Start() {
	var currentInput <-chan Job = p.input
	for i, stage := range p.stages {
		nextInput := make(chan Job, p.pool.size) // Buffered por stage
		p.wg.Add(1)
		go func(stageIdx int, s Stage, in <-chan Job, out chan<- Job) {
			defer p.wg.Done()
			for {
				select {
				case <-p.pool.ctx.Done():
					return
				case job, ok := <-in:
					if !ok {
						return
					}
					start := time.Now()
					processed, err := (&Worker{ID: stageIdx}).Process(p.pool.ctx, job, s)
					p.pool.metrics.throughput.Add(1)
					p.pool.metrics.latency.Add(int64(time.Since(start)))
					if err != nil {
						p.errs <- err // Manejo de errores sin detener pipeline
						continue
					}
					out <- processed
				}
			}
		}(i, stage, currentInput, nextInput)
		currentInput = nextInput
	}
	// Último stage envía a output
	p.wg.Add(1)
	go func() {
		defer p.wg.Done()
		for job := range currentInput {
			p.output <- job
		}
		close(p.output)
	}()
}
// Shutdown realiza cierre graceful esperando trabajos en curso.
func (p *Pipeline) Shutdown() {
	p.pool.cancel()
	p.wg.Wait()
	close(p.errs)
}
// Simulación de stages para procesamiento de imágenes.
func resizeStage(_ context.Context, j Job) (Job, error) {
	j.Data += "_resized"
	return j, nil
}
func filterStage(_ context.Context, j Job) (Job, error) {
	if j.ID%2 == 0 {
		return Job{}, errors.New("filter error")
	}
	j.Data += "_filtered"
	return j, nil
}
func saveStage(_ context.Context, j Job) (Job, error) {
	j.Data += "_saved"
	return j, nil
}
func main() {
	ctx := context.Background()
	pool := NewWorkerPool(5, ctx)
	input := make(chan Job, 1000)
	stages := []Stage{resizeStage, filterStage, saveStage}
	pipeline := ProcessPipeline(input, stages, pool)
	pipeline.Start()
	// Enviar 1000+ jobs
	go func() {
		for i := 0; i < 2000; i++ {
			input <- Job{ID: i, Data: fmt.Sprintf("image_%d", i)}
		}
		close(input)
	}()
	// Recibir outputs y errores
	var processed int
	for job := range pipeline.output {
		processed++
		// fmt.Println("Processed:", job)
	}
	for err := range pipeline.errs {
		// fmt.Println("Error:", err)
	}
	pipeline.Shutdown()
	// Métricas
	throughput := pool.metrics.throughput.Load()
	latency := pool.metrics.latency.Load() / throughput
	fmt.Printf("Processed %d jobs, throughput: %d, avg latency: %d ns\n", processed, throughput, latency)
	// Output esperado: Sistema que procesa 1000+ trabajos/segundo con múltiples stages, mostrando métricas de rendimiento y manejo elegante de errores
	// Ejemplo: Processed 1000 jobs, throughput: 1000, avg latency: X ns
}

// Sistema de Procesamiento de Pedidos E-commerce
// Implementación completa de un sistema de pedidos que combina Repository, Service y Circuit Breaker patterns para alta disponibilidad
package main
import (
	"context"
	"errors"
	"fmt"
	"log"
	"os"
	"sync"
	"sync/atomic"
	"time"
)
// Order representa un pedido.
type Order struct {
	ID        int
	Item      string
	Quantity  int
}
// Repository interface para backends.
type Repository interface {
	GetInventory(ctx context.Context, item string) (int, error)
	UpdateInventory(ctx context.Context, item string, quantity int) error
}
// InMemoryRepository implementación simple.
type InMemoryRepository struct {
	inventory map[string]int
	mu        sync.Mutex
}
func NewInMemoryRepository() *InMemoryRepository {
	return &InMemoryRepository{inventory: map[string]int{"item1": 100}}
}
func (r *InMemoryRepository) GetInventory(ctx context.Context, item string) (int, error) {
	r.mu.Lock()
	defer r.mu.Unlock()
	return r.inventory[item], nil
}
func (r *InMemoryRepository) UpdateInventory(ctx context.Context, item string, quantity int) error {
	r.mu.Lock()
	defer r.mu.Unlock()
	if r.inventory[item] < quantity {
		return errors.New("insufficient inventory")
	}
	r.inventory[item] -= quantity
	return nil
}
// CircuitBreaker para servicios externos.
type CircuitBreaker struct {
	state      string // open, closed, half-open
	failures   atomic.Int, threshold int
	timeout    time.Duration
	mu         sync.Mutex
	lastAttempt time.Time
}
func NewCircuitBreaker(threshold int, timeout time.Duration) *CircuitBreaker {
	return &CircuitBreaker{state: "closed", threshold: threshold, timeout: timeout}
}
func (cb *CircuitBreaker) Execute(fn func() error) error {
	cb.mu.Lock()
	if cb.state == "open" && time.Since(cb.lastAttempt) > cb.timeout {
		cb.state = "half-open"
	}
	if cb.state == "open" {
		cb.mu.Unlock()
		return errors.New("circuit open")
	}
	cb.mu.Unlock()
	err := fn()
	cb.mu.Lock()
	defer cb.mu.Unlock()
	if err != nil {
		cb.failures.Add(1)
		if cb.failures.Load() >= int32(cb.threshold) {
			cb.state = "open"
			cb.lastAttempt = time.Now()
		}
		return err
	}
	cb.state = "closed"
	cb.failures.Store(0)
	return nil
}
// PaymentService simulado con circuit breaker.
type PaymentService struct {
	cb *CircuitBreaker
}
func NewPaymentService() *PaymentService {
	return &PaymentService{cb: NewCircuitBreaker(3, 10*time.Second)}
}
func (s *PaymentService) ProcessPayment(ctx context.Context, order Order) error {
	return s.cb.Execute(func() error {
		// Simular fallo aleatorio
		if order.ID%2 == 0 {
			return errors.New("payment failed")
		}
		return nil
	})
}
// NotificationService similar.
type NotificationService struct {
	cb *CircuitBreaker
}
func NewNotificationService() *NotificationService {
	return &NotificationService{cb: NewCircuitBreaker(3, 10*time.Second)}
}
func (s *NotificationService) SendNotification(ctx context.Context, order Order) error {
	return s.cb.Execute(func() error {
		// Simular
		return nil
	})
}
// OrderService con inyección de dependencias.
type OrderService struct {
	repo       Repository
	payment    *PaymentService
	notify     *NotificationService
	metrics    struct {
		ordersPerMin atomic.Int64
		errorRate    atomic.Int64
		latency      atomic.Int64
	}
	logger     *log.Logger
}
func NewOrderService(repo Repository, payment *PaymentService, notify *NotificationService) *OrderService {
	logger := log.New(os.Stdout, "INFO: ", log.Ldate|log.Ltime)
	return &OrderService{repo: repo, payment: payment, notify: notify, logger: logger}
}
func (s *OrderService) ProcessOrder(ctx context.Context, order Order) error {
	start := time.Now()
	inv, err := s.repo.GetInventory(ctx, order.Item)
	if err != nil {
		s.metrics.errorRate.Add(1)
		s.logger.Printf("Error getting inventory: %v", err)
		return err
	}
	if inv < order.Quantity {
		s.metrics.errorRate.Add(1)
		return errors.New("insufficient inventory")
	}
	if err := s.payment.ProcessPayment(ctx, order); err != nil {
		s.metrics.errorRate.Add(1)
		s.logger.Printf("Payment failed: %v", err)
		return err // Circuit breaker maneja recuperación
	}
	if err := s.repo.UpdateInventory(ctx, order.Item, order.Quantity); err != nil {
		s.metrics.errorRate.Add(1)
		return err
	}
	if err := s.notify.SendNotification(ctx, order); err != nil {
		s.logger.Printf("Notification failed: %v", err)
		// No falla el pedido, solo log
	}
	s.metrics.ordersPerMin.Add(1)
	s.metrics.latency.Add(int64(time.Since(start)))
	return nil
}
func main() {
	ctx := context.Background()
	repo := NewInMemoryRepository()
	payment := NewPaymentService()
	notify := NewNotificationService()
	service := NewOrderService(repo, payment, notify)
	// Configuración via env (simulada)
	os.Setenv("THRESHOLD", "3")
	// Procesar 500+ pedidos concurrentes
	var wg sync.WaitGroup
	for i := 0; i < 600; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			order := Order{ID: id, Item: "item1", Quantity: 1}
			if err := service.ProcessOrder(ctx, order); err != nil {
				// fmt.Println("Error processing:", err)
			}
		}(i)
	}
	wg.Wait()
	// Métricas
	orders := service.metrics.ordersPerMin.Load()
	errors := service.metrics.errorRate.Load()
	latency := service.metrics.latency.Load() / orders
	fmt.Printf("Processed %d orders, error rate: %d, avg latency: %d ns\n", orders, errors, latency)
	// Output esperado: Sistema completo que maneja 500+ pedidos concurrentes, con recuperación automática ante fallos y métricas detalladas de rendimiento
	// Ejemplo: Processed 600 orders, error rate: X, avg latency: Y ns
}

// Ejemplos de Errores Comunes
// Error 1: Singleton con variables globales y race conditions
// Implementación incorrecta usando var global + sync.Once que causa dependencias ocultas
// Consecuencias: Tests no determinísticos, dificultad para mocking, acoplamiento excesivo
// Código incorrecto
package main
import (
	"fmt"
	"sync"
)
var globalInstance *Singleton
var once sync.Once
type Singleton struct {
	value int
}
func GetSingleton() *Singleton {
	once.Do(func() {
		globalInstance = &Singleton{value: 42}
	})
	return globalInstance
}
func main() {
	s1 := GetSingleton()
	s2 := GetSingleton()
	s1.value = 100 // Race condition si concurrente
	fmt.Println(s2.value) // Puede ser 42 o 100, no determinístico
}
// Solución: Usar dependency injection con interfaces y factory functions
// Esto permite mocking y evita globals
type SingletonInterface interface {
	GetValue() int
	SetValue(int)
}
type singletonImpl struct {
	value int
}
func (s *singletonImpl) GetValue() int { return s.value }
func (s *singletonImpl) SetValue(v int) { s.value = v }
func NewSingleton() SingletonInterface {
	return &singletonImpl{value: 42}
}
func UseSingleton(si SingletonInterface) {
	si.SetValue(100)
	fmt.Println(si.GetValue()) // Siempre determinístico
}
func main() {
	si := NewSingleton()
	UseSingleton(si)
}
// Error 2: Channels sin buffer causando deadlocks
// Uso de channels unbuffered en patrones producer-consumer sin sincronización adecuada
// Consecuencias: Goroutines bloqueadas permanentemente, deadlocks en runtime
// Código incorrecto
func incorrectChannel() {
	ch := make(chan int) // Unbuffered
	go func() {
		ch <- 1 // Bloquea si no hay receiver listo
	}()
	// No hay receive, deadlock
	<-ch
}
// Solución: Análisis de flujo de datos y uso correcto de buffered channels
// Usa buffer para evitar bloqueos
func correctChannel() {
	ch := make(chan int, 1) // Buffered
	go func() {
		ch <- 1 // No bloquea
		close(ch)
	}()
	val := <-ch
	fmt.Println(val) // Funciona sin deadlock
}
// Error 3: Ignorar context.Context en operaciones de larga duración
// Servicios que no propagan context para cancelación y timeouts
// Consecuencias: Memory leaks, operaciones zombie, imposibilidad de cancelar requests
// Código incorrecto
func longOperation() {
	time.Sleep(10 * time.Second) // No chequea context, sigue corriendo aun si cancelado
}
// Solución: Propagación correcta de context en toda la cadena de llamadas
// Chequea context para cancelación
func correctLongOperation(ctx context.Context) error {
	select {
	case <-time.After(10 * time.Second):
		return nil
	case <-ctx.Done():
		return ctx.Err() // Cancela si context done
	}
}
func main() {
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if err := correctLongOperation(ctx); err != nil {
		fmt.Println("Canceled:", err)
	}
	// Output: Canceled: context deadline exceeded
}