// Servidor HTTP Concurrente con Worker Pool
// Sistema que procesa requests HTTP usando un pool de workers para operaciones intensivas
package main
import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"syscall"
	"time"
)
// Task representa una tarea de procesamiento de archivo
type Task struct {
	FileName string
	// Simula datos de upload
	Data []byte
}
// startServer inicia el servidor HTTP con rutas configuradas
func startServer(addr string, tasks chan Task) {
	mux := http.NewServeMux()
	mux.HandleFunc("/upload", func(w http.ResponseWriter, r *http.Request) {
		processUpload(w, r, tasks)
	})
	srv := &http.Server{
		Addr:    addr,
		Handler: mux,
	}
	// Graceful shutdown con signal handling
	go func() {
		sigint := make(chan os.Signal, 1)
		signal.Notify(sigint, os.Interrupt, syscall.SIGTERM)
		<-sigint
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := srv.Shutdown(ctx); err != nil {
			log.Printf("HTTP server Shutdown: %v", err)
		}
	}()
	log.Printf("Starting server on %s", addr)
	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
		log.Fatalf("ListenAndServe(): %v", err)
	}
}
// processUpload maneja el upload y envía trabajo al pool
func processUpload(w http.ResponseWriter, r *http.Request, tasks chan Task) {
	if r.Method != http.MethodPost {
		http.Error(w, "Invalid request method", http.StatusMethodNotAllowed)
		return
	}
	// Simula parsing de upload (en producción usar multipart)
	fileName := r.FormValue("filename")
	if fileName == "" {
		http.Error(w, "Missing filename", http.StatusBadRequest)
		return
	}
	// Envía tarea al pool con context y timeout
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	select {
	case tasks <- Task{FileName: fileName, Data: []byte("simulated data")}:
		fmt.Fprintf(w, "Upload queued: %s\n", fileName)
	case <-ctx.Done():
		http.Error(w, "Timeout queuing task", http.StatusGatewayTimeout)
	}
}
// workerPool crea un pool de workers que procesan tareas concurrentemente
func workerPool(numWorkers int, tasks <-chan Task) {
	for i := 0; i < numWorkers; i++ {
		go func(id int) {
			for task := range tasks {
				processFile(task, id)
			}
		}(i)
	}
}
// processFile simula procesamiento intensivo
func processFile(task Task, workerID int) {
	// Simula trabajo intensivo
	time.Sleep(2 * time.Second)
	log.Printf("Worker %d processed file: %s", workerID, task.FileName)
}
func main() {
	tasks := make(chan Task, 10) // Channel buffered para worker pool
	workerPool(5, tasks)         // Inicia pool con 5 workers
	startServer(":8080", tasks)
}
// Output esperado como comentarios:
// Servidor que maneja múltiples requests simultáneos con procesamiento asíncrono y logs de rendimiento
// Ejemplo de log: Worker 1 processed file: example.jpg
// Servidor inicia en :8080 y maneja shutdown graceful

// Sistema de Procesamiento de Imágenes Distribuido
// Servicio de producción para redimensionar y optimizar imágenes con múltiples workers
package main
import (
	"context"
	"fmt"
	"log"
	"net/http"
	"os"
	"os/signal"
	"sync"
	"syscall"
	"time"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promhttp"
	"golang.org/x/time/rate"
)
// ImageTask representa una tarea de procesamiento de imagen
type ImageTask struct {
	FileName string
	Size     string // e.g., "thumbnail", "full"
	Data     []byte
}
// Configuración para diferentes tamaños de imagen
var imageSizes = map[string]struct {
	Width  int
	Height int
}{
	"thumbnail": {Width: 100, Height: 100},
	"full":      {Width: 800, Height: 600},
}
// Métricas de rendimiento con Prometheus
var (
	processedImages = prometheus.NewCounterVec(
		prometheus.CounterOpts{
			Name: "processed_images_total",
			Help: "Total number of images processed",
		},
		[]string{"size"},
	)
	latencyHistogram = prometheus.NewHistogramVec(
		prometheus.HistogramOpts{
			Name:    "image_processing_latency_seconds",
			Help:    "Latency of image processing",
			Buckets: prometheus.DefBuckets,
		},
		[]string{"size"},
	)
)
func init() {
	prometheus.MustRegister(processedImages)
	prometheus.MustRegister(latencyHistogram)
}
// startServer inicia el servidor con métricas y rutas
func startServer(addr string, tasks chan ImageTask) {
	mux := http.NewServeMux()
	mux.HandleFunc("/upload", func(w http.ResponseWriter, r *http.Request) {
		processImageUpload(w, r, tasks)
	})
	mux.Handle("/metrics", promhttp.Handler())
	srv := &http.Server{
		Addr:    addr,
		Handler: mux,
	}
	go func() {
		sigint := make(chan os.Signal, 1)
		signal.Notify(sigint, os.Interrupt, syscall.SIGTERM)
		<-sigint
		ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
		defer cancel()
		if err := srv.Shutdown(ctx); err != nil {
			log.Printf("HTTP server Shutdown: %v", err)
		}
	}()
	log.Printf("Starting server on %s", addr)
	if err := srv.ListenAndServe(); err != http.ErrServerClosed {
		log.Fatalf("ListenAndServe(): %v", err)
	}
}
// Rate limiter y circuit breaker simple
var limiter = rate.NewLimiter(rate.Every(time.Second), 100) // 100 req/s
// Circuit breaker state
var (
	cbOpen     bool
	cbMu       sync.Mutex
	cbFailures int
	cbThreshold = 5
)
// processImageUpload maneja upload con rate limiting y circuit breaker
func processImageUpload(w http.ResponseWriter, r *http.Request, tasks chan ImageTask) {
	if r.Method != http.MethodPost {
		http.Error(w, "Invalid method", http.StatusMethodNotAllowed)
		return
	}
	cbMu.Lock()
	if cbOpen {
		cbMu.Unlock()
		http.Error(w, "Circuit open", http.StatusServiceUnavailable)
		return
	}
	cbMu.Unlock()
	if err := limiter.Wait(r.Context()); err != nil {
		http.Error(w, "Rate limit exceeded", http.StatusTooManyRequests)
		return
	}
	fileName := r.FormValue("filename")
	size := r.FormValue("size")
	if fileName == "" || size == "" || imageSizes[size].Width == 0 {
		http.Error(w, "Invalid params", http.StatusBadRequest)
		return
	}
	ctx, cancel := context.WithTimeout(r.Context(), 10*time.Second)
	defer cancel()
	select {
	case tasks <- ImageTask{FileName: fileName, Size: size, Data: []byte("simulated image")}:
		fmt.Fprintf(w, "Image queued: %s size %s\n", fileName, size)
	case <-ctx.Done():
		cbMu.Lock()
		cbFailures++
		if cbFailures >= cbThreshold {
			cbOpen = true
			go func() {
				time.Sleep(30 * time.Second)
				cbMu.Lock()
				cbOpen = false
				cbFailures = 0
				cbMu.Unlock()
			}()
		}
		cbMu.Unlock()
		http.Error(w, "Timeout", http.StatusGatewayTimeout)
	}
}
// workerPool con fan-out/fan-in pattern
func workerPool(numWorkers int, tasks <-chan ImageTask, results chan<- string) {
	var wg sync.WaitGroup
	for i := 0; i < numWorkers; i++ {
		wg.Add(1)
		go func(id int) {
			defer wg.Done()
			for task := range tasks {
				result := processImage(task, id)
				results <- result
			}
		}(i)
	}
	// Fan-in: recolecta resultados
	go func() {
		wg.Wait()
		close(results)
	}()
}
// processImage simula procesamiento con métricas
func processImage(task ImageTask, workerID int) string {
	start := time.Now()
	// Simula redimensionamiento
	time.Sleep(500 * time.Millisecond) // Simula trabajo
	duration := time.Since(start).Seconds()
	processedImages.WithLabelValues(task.Size).Inc()
	latencyHistogram.WithLabelValues(task.Size).Observe(duration)
	log.Printf("Worker %d processed %s image: %s in %f s", workerID, task.Size, task.FileName, duration)
	return fmt.Sprintf("Processed: %s", task.FileName)
}
func main() {
	tasks := make(chan ImageTask, 100) // Buffered channel for fan-out
	results := make(chan string, 100)  // For fan-in
	workerPool(10, tasks, results)
	// Recolecta resultados (fan-in)
	go func() {
		for result := range results {
			log.Println(result)
		}
	}()
	startServer(":8080", tasks)
}
// Output esperado como comentarios:
// Sistema que procesa 1000+ imágenes/minuto con métricas de latencia y throughput
// Métricas expuestas en /metrics, e.g., processed_images_total{size="thumbnail"} 100
// Logs como: Worker 1 processed thumbnail image: img.jpg in 0.5 s

// Errores Comunes en Sistemas Concurrentes
// Error 1: Goroutine Leak por Channel Bloqueado
// Descripción: Goroutine que envía a channel sin receptor, quedando bloqueada permanentemente
// Consecuencias: Acumulación de memoria y degradación del rendimiento
// Código que demuestra el error (incorrecto)
package main
import (
	"fmt"
	"time"
)
func main() {
	ch := make(chan int) // Channel no buffered
	// Goroutine que intenta enviar sin receptor
	go func() {
		ch <- 42 // Se bloquea indefinidamente si no hay receptor
		fmt.Println("Sent") // Nunca se ejecuta
	}()
	time.Sleep(1 * time.Second)
	fmt.Println("Main ended") // Ejecuta, pero goroutine leak
}
// Solución: Usar select con timeout o context para evitar bloqueos indefinidos
// Código corregido
func correctedLeak() {
	ch := make(chan int)
	go func() {
		select {
		case ch <- 42:
			fmt.Println("Sent")
		case <-time.After(500 * time.Millisecond):
			fmt.Println("Timeout, avoiding leak")
		}
	}()
	time.Sleep(1 * time.Second)
	fmt.Println("Main ended")
}
// Error 2: Race Condition en Contador Compartido
// Descripción: Múltiples goroutines modificando variable sin sincronización
// Consecuencias: Resultados inconsistentes y posibles crashes
// Código que demuestra el error (incorrecto)
func raceConditionIncorrect() {
	var counter int
	var wg sync.WaitGroup
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			counter++ // Race condition: acceso concurrente sin protección
		}()
	}
	wg.Wait()
	fmt.Printf("Counter: %d\n", counter) // Puede no ser 1000 debido a race
}
// Solución: Usar atomic operations o mutex para proteger acceso concurrente
// Código corregido con atomic
import (
	"sync/atomic"
)
func raceConditionCorrected() {
	var counter int32
	var wg sync.WaitGroup
	for i := 0; i < 1000; i++ {
		wg.Add(1)
		go func() {
			defer wg.Done()
			atomic.AddInt32(&counter, 1) // Acceso atómico seguro
		}()
	}
	wg.Wait()
	fmt.Printf("Counter: %d\n", counter) // Siempre 1000
}
// Output esperado como comentarios:
// En incorrecto: Posible output Counter: 998 (inconsistente)
// En corregido: Counter: 1000 (consistente)