Go Concurrency a Escala: Procesando 100,000 Registros Eficientemente con REST y DDD

La Realidad: Procesamiento de Datos Empresariales

Las organizaciones procesan datasets masivos diariamente. Los bancos ingieren millones de transacciones. Las plataformas logísticas rastrean millones de envíos. Las autoridades fiscales validan miles de millones de documentos.

El problema no es si puedes procesar 100,000 registros. El problema es hacerlo en menos de 20 minutos mientras mantienes consistencia, recuperación de errores y auditoría.

Esto no es arquitectura teórica. Esto es lo que los sistemas de producción demandan.

Parte 1: Fundamento de Diseño Orientado al Dominio

Antes de escribir código de concurrencia, establece el modelo de dominio.

El Dominio Fiscal

En sistemas fiscales mexicanos, necesitas procesar:

Documento Fiscal
├── UUID (identificador único)
├── InvoiceData (el hecho de negocio)
├── ComplianceStatus (validado contra reglas SAT)
├── AuditTrail (cuándo, por quién, qué cambió)
└── StorageLocation (dónde vive el XML)

Define esto en código:

// domain/fiscal.go
package domain

import "time"

// InvoiceStatus representa estados de cumplimiento fiscal
type InvoiceStatus string

const (
	StatusPending    InvoiceStatus = "pending"
	StatusValidated  InvoiceStatus = "validated"
	StatusRejected   InvoiceStatus = "rejected"
	StatusProcessing InvoiceStatus = "processing"
	StatusArchived   InvoiceStatus = "archived"
)

// FiscalRecord es la entidad de dominio central
type FiscalRecord struct {
	UUID              string           `json:"uuid"`
	InvoiceID         string           `json:"invoice_id"`
	Issuer            string           `json:"issuer_rfc"`
	Receiver          string           `json:"receiver_rfc"`
	Amount            decimal.Decimal  `json:"amount"`
	Currency          string           `json:"currency"`
	IssuanceDate      time.Time        `json:"issuance_date"`
	Status            InvoiceStatus    `json:"status"`
	FileURL           string           `json:"file_url"`
	Checksum          string           `json:"checksum"`
	CompressedSize    int64            `json:"compressed_size"`
	ProcessedAt       time.Time        `json:"processed_at,omitempty"`
	ErrorMessage      string           `json:"error_message,omitempty"`
	RetryCount        int              `json:"retry_count"`
	LastRetryAt       time.Time        `json:"last_retry_at,omitempty"`
}

// InvoiceLineItem representa línea de transacción individual
type InvoiceLineItem struct {
	RecordUUID  string
	Description string
	Quantity    decimal.Decimal
	UnitPrice   decimal.Decimal
	Amount      decimal.Decimal
	TaxRate     decimal.Decimal
	TaxAmount   decimal.Decimal
}

// Repository define contrato de acceso a datos
type FiscalRepository interface {
	SaveRecord(ctx context.Context, record *FiscalRecord) error
	FindByUUID(ctx context.Context, uuid string) (*FiscalRecord, error)
	FindByStatus(ctx context.Context, status InvoiceStatus, limit int) ([]*FiscalRecord, error)
	UpdateStatus(ctx context.Context, uuid string, status InvoiceStatus) error
	BulkInsert(ctx context.Context, records []*FiscalRecord) error
}

// UseCase define operaciones de negocio
type ProcessFiscalDocumentUseCase interface {
	Execute(ctx context.Context, uuid string) (*FiscalRecord, error)
}

// Event representa eventos de dominio para auditoría
type DomainEvent interface {
	AggregateID() string
	EventType() string
	OccurredAt() time.Time
}

type RecordProcessedEvent struct {
	UUID        string
	Status      InvoiceStatus
	ProcessedAt time.Time
	Duration    time.Duration
}

func (e *RecordProcessedEvent) AggregateID() string { return e.UUID }
func (e *RecordProcessedEvent) EventType() string    { return "fiscal.record.processed" }
func (e *RecordProcessedEvent) OccurredAt() time.Time { return e.ProcessedAt }

Esta base asegura que tu código refleje reglas de negocio, no solo plomería técnica.

Parte 2: Diagrama de Arquitectura — Visión General del Sistema

Entiende el flujo completo antes de implementar piezas:

graph TB
    subgraph "Client Layer"
        A["REST API Client<br/>(Admin, Analyst)"]
    end

    subgraph "API Gateway"
        B["HTTP Router<br/>(Chi)"]
    end

    subgraph "Application Layer"
        C["ProcessingController<br/>validateInput"]
        D["QueryController<br/>getStatus"]
        E["ReportController<br/>aggregateData"]
    end

    subgraph "Pipeline Orchestrator"
        F["PipelineOrchestrator<br/>coordinates stages"]
    end

    subgraph "Domain Layer"
        G1["QueryWorker<br/>100K UUIDs"]
        G2["DownloadWorker<br/>50K files"]
        G3["ExtractorWorker<br/>XML parsing"]
        G4["CompressorWorker<br/>gzip"]
        G5["IndexerWorker<br/>cache warm"]
    end

    subgraph "Infrastructure Layer"
        H1["PostgreSQL<br/>fiscal_records"]
        H2["SFTP Pool<br/>5-10 connections"]
        H3["Memory Cache<br/>RWMutex"]
        H4["S3/Disk<br/>compressed files"]
        H5["Event Log<br/>audit trail"]
    end

    subgraph "External Systems"
        I["SAT Server<br/>validation"]
    end

    A -->|POST /process<br/>GET /status| B
    B --> C
    B --> D
    B --> E
    C --> F
    F -->|fan-out| G1
    F -->|fan-out| G2
    F -->|fan-out| G3
    F -->|fan-out| G4
    F -->|fan-out| G5

    G1 -->|queries| H1
    G2 -->|downloads| H2
    G3 -->|updates| H1
    G4 -->|stores| H4
    G5 -->|indexes| H3
    G4 -->|events| H5
    G3 -.->|validate| I

    H3 -->|read O(1)| D
    H1 -->|read| E
    H5 -->|audit| E

    style F fill:#ff9800,stroke:#333,stroke-width:2px
    style H3 fill:#4caf50,stroke:#333,stroke-width:2px
    style A fill:#2196f3,stroke:#333,stroke-width:2px

Este diagrama muestra datos fluyendo a través de etapas independientes. Cada etapa es una responsabilidad de dominio.

Parte 3: Casos de Uso — Escenarios del Mundo Real

Caso de Uso 1: Ingesta de Facturas Fiscales (SAT México)

Escenario: Autoridad fiscal recibe 50,000 facturas diarias de negocios. Cada factura debe ser:

Descargada desde servidor SFTP del emisor
Validada contra reglas SAT
Almacenada con rastro de auditoría completo
Disponible para analistas en menos de 5 minutos

Requerimientos:

Procesamiento exactly-once (sin duplicados)
Los registros fallidos deben reintentar automáticamente
Validación de checksum antes de almacenamiento
Event sourcing para auditorías fiscales

Flujo de Procesamiento:

sequenceDiagram
    participant Cliente
    participant API
    participant Pipeline
    participant Query
    participant Download
    participant Validate
    participant Store

    Cliente->>API: POST /fiscal/process?date=2026-03-04
    API->>Pipeline: orchestrate(50000 UUIDs)

    par Query Stage
        Pipeline->>Query: semaphore(15)
        Query->>Query: SELECT * FROM daily_records
        Query-->>Pipeline: stream 50,000 records
    and Download Stage
        Pipeline->>Download: workers(25)
        Download->>Download: SFTP batch downloads
        Download-->>Pipeline: files in memory
    and Validate Stage
        Pipeline->>Validate: workers(20)
        Validate->>Validate: XML schema validation
        Validate->>Validate: SAT rule check
        Validate-->>Pipeline: validated data
    and Store Stage
        Pipeline->>Store: workers(10)
        Store->>Store: gzip compression
        Store->>Store: S3 upload
        Store-->>Pipeline: metadata
    end

    Pipeline-->>API: results (success/error count)
    API-->>Cliente: { processed: 49,500, failed: 500 }

Rendimiento Esperado:

Query stage: 67 segundos (100,000 ÷ 15 workers)
Download stage: 1,000 segundos (50,000 archivos × 500ms ÷ 25 workers)
Validation: 400 segundos (solapado con download)
Storage: 200 segundos (solapado con validation)
Total: 20 minutos (concurrencia solapada)

Caso de Uso 2: Sincronización de Datos Logísticos Transfronterizo

Escenario: Empresa logística internacional sincroniza datos de envíos desde 10 servidores SFTP regionales. Procesa 150,000 registros, extrae datos de rastreo, valida contra reglas aduanales, y actualiza base de datos central.

Complejidad: Multi-servidor, con pooling de conexiones por-servidor y reglas de validación regionales.

Arquitectura:

// usecase/sync_logistics.go
package usecase

type SyncRegionalShipmentsUseCase struct {
	repositories *Repositories
	sftp         map[string]*SFTPPool  // regional pools
	pipelines    map[string]*Pipeline  // per-region pipelines
	eventBus     EventBus
}

func (uc *SyncRegionalShipmentsUseCase) Execute(ctx context.Context, regions []string) (*SyncResult, error) {
	results := &SyncResult{
		Regions: make(map[string]*RegionResult),
	}

	// Procesar regiones en paralelo, cada una con pool SFTP dedicado
	for _, region := range regions {
		go func(r string) {
			pipeline := uc.pipelines[r]
			regionResult := pipeline.Process(ctx, uc.getRegionalRecords(ctx, r))
			results.Regions[r] = regionResult
		}(region)
	}

	return results, nil
}

Flujo de Datos Multi-Región:

graph LR
    subgraph "SFTP Pools"
        S1["Pool: US<br/>10 connections"]
        S2["Pool: EU<br/>10 connections"]
        S3["Pool: APAC<br/>10 connections"]
    end

    subgraph "Pipelines (Parallel)"
        P1["US Pipeline<br/>30 workers"]
        P2["EU Pipeline<br/>30 workers"]
        P3["APAC Pipeline<br/>30 workers"]
    end

    subgraph "Central Processing"
        CP["Merger Stage<br/>deduplicate<br/>cross-check"]
    end

    subgraph "Output"
        DB[(PostgreSQL<br/>unified)]
        AUDIT["Audit Log<br/>per-region"]
    end

    S1 -->|regional data| P1
    S2 -->|regional data| P2
    S3 -->|regional data| P3

    P1 -->|stream results| CP
    P2 -->|stream results| CP
    P3 -->|stream results| CP

    CP -->|merged| DB
    CP -->|events| AUDIT

Caso de Uso 3: Generación de Reportes en Lote con Cache

Escenario: Analistas ejecutan reportes diarios sobre 100,000 registros procesados. Reportes deben incluir:

Estadísticas de resumen (agregadas)
Desglose por categoría (indexado)
Análisis de tendencias (ventana rodante)
Actualizaciones en tiempo real conforme llega nuevo data

Solución: Calentador de cache durante completación de pipeline, sirver reportes desde memoria con lookups O(1).

// infrastructure/cache.go
package infrastructure

type IndexedFiscalCache struct {
	// Índices primarios
	byUUID       map[string]*FiscalRecord
	byIssuer     map[string][]*FiscalRecord
	byStatus     map[string][]*FiscalRecord
	byDateRange  map[string][]*FiscalRecord

	// Agregaciones (pre-computadas)
	totalByIssuer    map[string]decimal.Decimal
	countByStatus    map[string]int64
	dailyTotals      map[string]decimal.Decimal
	hourlyTotals     map[string]decimal.Decimal

	// Estadísticas
	mu sync.RWMutex
	lastUpdate time.Time
	recordCount int64
}

// WarmCache es llamado después de completación del pipeline
func (c *IndexedFiscalCache) WarmCache(ctx context.Context, repo FiscalRepository) error {
	c.mu.Lock()
	defer c.mu.Unlock()

	// Stream desde repository para evitar spike de memoria
	records, err := repo.FindByStatus(ctx, StatusValidated, 100000)
	if err != nil {
		return err
	}

	for _, record := range records {
		// Indexar por UUID
		c.byUUID[record.UUID] = record

		// Indexar por Issuer
		c.byIssuer[record.Issuer] = append(c.byIssuer[record.Issuer], record)

		// Agregar totales
		c.totalByIssuer[record.Issuer] = c.totalByIssuer[record.Issuer].Add(record.Amount)

		// Índices basados en tiempo
		dateKey := record.IssuanceDate.Format("2006-01-02")
		c.byDateRange[dateKey] = append(c.byDateRange[dateKey], record)
		c.dailyTotals[dateKey] = c.dailyTotals[dateKey].Add(record.Amount)
	}

	c.lastUpdate = time.Now()
	c.recordCount = int64(len(records))

	return nil
}

// GetDailyReport devuelve datos pre-agregados en O(1)
func (c *IndexedFiscalCache) GetDailyReport(date string) *DailyReport {
	c.mu.RLock()
	defer c.mu.RUnlock()

	return &DailyReport{
		Date:        date,
		RecordCount: int64(len(c.byDateRange[date])),
		TotalAmount: c.dailyTotals[date],
		ByIssuer:    c.getIssuerSummary(date),
		LastUpdate:  c.lastUpdate,
	}
}

Parte 4: Diseño de API REST — Procesamiento como Servicio

Define contratos claros para clientes que disparan operaciones masivas:

// api/controller.go
package api

import (
	"context"
	"encoding/json"
	"net/http"
	"time"

	"chi"
)

type ProcessingController struct {
	processingUC ProcessFiscalBatchUseCase
	repo         FiscalRepository
	eventBus     EventBus
}

// ProcessingRequest define qué un cliente envía
type ProcessingRequest struct {
	Source          string    `json:"source" validate:"required"`     // "sftp", "s3", "api"
	FilePattern     string    `json:"file_pattern" validate:"required"` // "*.xml"
	RecordIDs       []string  `json:"record_ids"`                       // UUIDs específicos a procesar
	ValidationRules string    `json:"validation_rules"`                 // "strict", "lenient"
	Priority        string    `json:"priority" validate:"required"`     // "normal", "high"
	CallbackURL     string    `json:"callback_url"`                     // webhook para completación
	MaxRetries      int       `json:"max_retries" validate:"min=0"`
}

// ProcessingResponse devuelto inmediatamente
type ProcessingResponse struct {
	JobID             string                 `json:"job_id"`
	Status            string                 `json:"status"`
	SubmittedAt       time.Time              `json:"submitted_at"`
	EstimatedDuration time.Duration          `json:"estimated_duration"`
	RecordCount       int                    `json:"record_count"`
	Metrics           map[string]interface{} `json:"metrics"`
}

// POST /api/v1/fiscal/process
func (pc *ProcessingController) StartProcessing(w http.ResponseWriter, r *http.Request) {
	ctx := r.Context()

	var req ProcessingRequest
	if err := json.NewDecoder(r.Body).Decode(&req); err != nil {
		http.Error(w, "invalid request", http.StatusBadRequest)
		return
	}

	// Validar solicitud
	if err := pc.validateRequest(req); err != nil {
		http.Error(w, err.Error(), http.StatusBadRequest)
		return
	}

	// Crear trabajo de procesamiento
	job := &ProcessingJob{
		ID:              generateJobID(),
		Source:          req.Source,
		FilePattern:     req.FilePattern,
		RecordIDs:       req.RecordIDs,
		ValidationRules: req.ValidationRules,
		Priority:        req.Priority,
		Status:          JobStatusQueued,
		SubmittedAt:     time.Now(),
		MaxRetries:      req.MaxRetries,
		CallbackURL:     req.CallbackURL,
	}

	// Encolar trabajo (async)
	go pc.processingUC.Execute(context.Background(), job)

	// Devolver inmediatamente
	response := ProcessingResponse{
		JobID:             job.ID,
		Status:            JobStatusQueued,
		SubmittedAt:       job.SubmittedAt,
		EstimatedDuration: pc.estimateDuration(req.RecordIDs),
		RecordCount:       len(req.RecordIDs),
		Metrics: map[string]interface{}{
			"queue_depth": pc.processingUC.QueueDepth(),
			"workers":     pc.processingUC.WorkerCount(),
		},
	}

	w.Header().Set("Content-Type", "application/json")
	w.Header().Set("Location", fmt.Sprintf("/api/v1/jobs/%s", job.ID))
	w.WriteHeader(http.StatusAccepted)
	json.NewEncoder(w).Encode(response)
}

// GET /api/v1/jobs/:jobId
func (pc *ProcessingController) GetJobStatus(w http.ResponseWriter, r *http.Request) {
	jobID := chi.URLParam(r, "jobId")

	job, err := pc.processingUC.GetJob(r.Context(), jobID)
	if err != nil {
		http.Error(w, "job not found", http.StatusNotFound)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(job)
}

// GET /api/v1/fiscal/records?status=validated&limit=100
func (pc *ProcessingController) ListRecords(w http.ResponseWriter, r *http.Request) {
	status := r.URL.Query().Get("status")
	limit := 100

	records, err := pc.repo.FindByStatus(r.Context(), status, limit)
	if err != nil {
		http.Error(w, err.Error(), http.StatusInternalServerError)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(map[string]interface{}{
		"count":   len(records),
		"records": records,
	})
}

// GET /api/v1/reports/daily?date=2026-03-04
func (pc *ProcessingController) GetDailyReport(w http.ResponseWriter, r *http.Request) {
	date := r.URL.Query().Get("date")

	report := pc.cache.GetDailyReport(date)
	if report == nil {
		http.Error(w, "no data for date", http.StatusNotFound)
		return
	}

	w.Header().Set("Content-Type", "application/json")
	json.NewEncoder(w).Encode(report)
}

func (pc *ProcessingController) estimateDuration(recordIDs []string) time.Duration {
	// Estimación aproximada: 50,000 registros / 20 workers = 2500 rondas
	// Cada ronda ~500ms (descarga) + 50ms (extracción) + 25ms (compresión)
	// Total ~575ms * 2500 = ~1,437 segundos ≈ 24 minutos
	return time.Duration(len(recordIDs)/20) * 575 * time.Millisecond
}

Flujo de API:

sequenceDiagram
    participant Cliente
    participant API
    participant JobQueue
    participant Pipeline
    participant Cache
    participant Webhook

    Cliente->>API: POST /api/v1/fiscal/process<br/>{recordIds: [uuid1, uuid2, ...]}
    API->>JobQueue: enqueue(ProcessingJob)
    API-->>Cliente: 202 Accepted<br/>{jobId: "job_123"}

    Pipeline->>Pipeline: Process records<br/>(20 minutes)
    Pipeline->>Cache: WarmCache()
    Pipeline->>Webhook: POST callback<br/>{status: "completed"}

    par Polling
        Cliente->>API: GET /api/v1/jobs/job_123
        API-->>Cliente: {status: "processing", progress: 45%}
    and Immediate Query
        Cliente->>API: GET /api/v1/reports/daily?date=2026-03-04
        API->>Cache: query O(1)
        API-->>Cliente: {totalAmount: 1500000, count: 49500}
    end

Parte 5: Implementación de Pipeline con Eventos de Dominio

// application/pipeline.go
package application

type PipelineStage interface {
	Process(ctx context.Context, input <-chan interface{}) <-chan interface{}
	Name() string
	WorkerCount() int
}

type FiscalProcessingPipeline struct {
	stages       []PipelineStage
	eventBus     EventBus
	errorHandler ErrorHandler
	metrics      MetricsCollector
}

func NewFiscalProcessingPipeline(eventBus EventBus, metrics MetricsCollector) *FiscalProcessingPipeline {
	return &FiscalProcessingPipeline{
		stages: []PipelineStage{
			NewQueryStage(15),           // 15 consultas DB concurrentes
			NewDownloadStage(25),        // 25 descargas SFTP concurrentes
			NewValidationStage(20),      // 20 validaciones concurrentes
			NewCompressionStage(10),     // 10 operaciones gzip concurrentes
			NewStorageStage(5),          // 5 uploads S3 concurrentes
			NewIndexingStage(1),         // 1 indexador para calentamiento de cache
		},
		eventBus:     eventBus,
		errorHandler: NewErrorHandler(),
		metrics:      metrics,
	}
}

// Execute corre el pipeline completo
func (p *FiscalProcessingPipeline) Execute(ctx context.Context, job *ProcessingJob) (*ProcessingResult, error) {
	startTime := time.Now()
	defer func() {
		p.metrics.RecordPipelineExecution(time.Since(startTime))
	}()

	// Publicar evento de inicio
	p.eventBus.Publish(&JobStartedEvent{
		JobID:     job.ID,
		StartedAt: startTime,
	})

	// Encadenar etapas: output de etapa N → input de etapa N+1
	input := p.createInput(ctx, job.RecordIDs)

	var stageOutput <-chan interface{}
	for _, stage := range p.stages {
		stageMetrics := p.metrics.BeginStage(stage.Name(), stage.WorkerCount())

		stageOutput = stage.Process(ctx, input)
		input = stageOutput

		stageMetrics.Complete()
	}

	// Recopilar resultados y errores
	result := &ProcessingResult{
		JobID:       job.ID,
		StartedAt:   startTime,
		Successful:  0,
		Failed:      0,
		Errors:      []error{},
	}

	for item := range stageOutput {
		if record, ok := item.(*domain.FiscalRecord); ok {
			if record.Status == domain.StatusArchived {
				result.Successful++
			} else if record.Status == domain.StatusRejected {
				result.Failed++
				result.Errors = append(result.Errors, errors.New(record.ErrorMessage))
			}
		}
	}

	result.CompletedAt = time.Now()
	result.Duration = result.CompletedAt.Sub(startTime)

	// Publicar evento de completación
	p.eventBus.Publish(&JobCompletedEvent{
		JobID:       job.ID,
		Successful:  result.Successful,
		Failed:      result.Failed,
		Duration:    result.Duration,
		CompletedAt: result.CompletedAt,
	})

	return result, nil
}

func (p *FiscalProcessingPipeline) createInput(ctx context.Context, recordIDs []string) <-chan interface{} {
	out := make(chan interface{}, 100)

	go func() {
		defer close(out)
		for _, id := range recordIDs {
			select {
			case <-ctx.Done():
				return
			case out <- id:
			}
		}
	}()

	return out
}

Parte 6: Etapa de Consulta — Acceso Concurrente a Base de Datos

// domain/stage_query.go
package domain

type QueryStage struct {
	db      *sql.DB
	workers int
	out     chan interface{}
}

func NewQueryStage(workers int) *QueryStage {
	return &QueryStage{
		workers: workers,
		out:     make(chan interface{}, 100),
	}
}

func (qs *QueryStage) Process(ctx context.Context, input <-chan interface{}) <-chan interface{} {
	sem := make(chan struct{}, qs.workers)
	var wg sync.WaitGroup

	go func() {
		for uuid := range input {
			wg.Add(1)
			go func(id interface{}) {
				defer wg.Done()

				// Adquirir semáforo
				select {
				case sem <- struct{}{}:
					defer func() { <-sem }()
				case <-ctx.Done():
					return
				}

				// Consultar base de datos
				record, err := qs.queryFiscalRecord(ctx, id.(string))
				if err != nil {
					qs.out <- &FiscalRecord{UUID: id.(string), ErrorMessage: err.Error()}
					return
				}

				qs.out <- record
			}(uuid)
		}

		wg.Wait()
		close(qs.out)
	}()

	return qs.out
}

func (qs *QueryStage) queryFiscalRecord(ctx context.Context, uuid string) (*FiscalRecord, error) {
	row := qs.db.QueryRowContext(ctx, `
		SELECT uuid, invoice_id, issuer, receiver, amount, currency,
		       issuance_date, status, file_url, checksum, retry_count
		FROM fiscal_records
		WHERE uuid = $1
	`, uuid)

	var record FiscalRecord
	err := row.Scan(
		&record.UUID, &record.InvoiceID, &record.Issuer, &record.Receiver,
		&record.Amount, &record.Currency, &record.IssuanceDate, &record.Status,
		&record.FileURL, &record.Checksum, &record.RetryCount,
	)

	return &record, err
}

func (qs *QueryStage) Name() string      { return "Query" }
func (qs *QueryStage) WorkerCount() int  { return qs.workers }

Parte 7: Etapa de Descarga con Pooling de Conexiones SFTP

// infrastructure/sftp_stage.go
package infrastructure

type DownloadStage struct {
	pool    *SFTPPool
	workers int
	out     chan interface{}
}

func NewDownloadStage(workers int) *DownloadStage {
	return &DownloadStage{
		workers: workers,
		out:     make(chan interface{}, 100),
	}
}

func (ds *DownloadStage) Process(ctx context.Context, input <-chan interface{}) <-chan interface{} {
	var wg sync.WaitGroup

	// Iniciar N workers
	for i := 0; i < ds.workers; i++ {
		wg.Add(1)
		go ds.downloadWorker(ctx, input, &wg)
	}

	go func() {
		wg.Wait()
		close(ds.out)
	}()

	return ds.out
}

func (ds *DownloadStage) downloadWorker(ctx context.Context, input <-chan interface{}, wg *sync.WaitGroup) {
	defer wg.Done()

	for record := range input {
		select {
		case <-ctx.Done():
			return
		default:
		}

		fiscal := record.(*FiscalRecord)

		// Obtener conexión del pool
		conn, err := ds.pool.Get()
		if err != nil {
			fiscal.Status = StatusRejected
			fiscal.ErrorMessage = fmt.Sprintf("connection pool: %v", err)
			ds.out <- fiscal
			continue
		}

		// Descargar y verificar hash
		data, checksum, err := ds.downloadAndVerify(conn, fiscal.FileURL)
		ds.pool.Return(conn)

		if err != nil {
			fiscal.Status = StatusRejected
			fiscal.ErrorMessage = fmt.Sprintf("download failed: %v", err)
			fiscal.RetryCount++
		} else if checksum != fiscal.Checksum {
			fiscal.Status = StatusRejected
			fiscal.ErrorMessage = fmt.Sprintf("checksum mismatch: expected %s, got %s", fiscal.Checksum, checksum)
		} else {
			fiscal.Status = StatusDownloaded
			fiscal.RawXMLData = data
		}

		ds.out <- fiscal
	}
}

func (ds *DownloadStage) downloadAndVerify(conn *sftp.Client, remotePath string) ([]byte, string, error) {
	file, err := conn.Open(remotePath)
	if err != nil {
		return nil, "", err
	}
	defer file.Close()

	// Stream download con cálculo de hash
	var buf bytes.Buffer
	h := sha256.New()
	writer := io.MultiWriter(&buf, h)

	if _, err := io.Copy(writer, file); err != nil {
		return nil, "", err
	}

	checksum := fmt.Sprintf("%x", h.Sum(nil))
	return buf.Bytes(), checksum, nil
}

func (ds *DownloadStage) Name() string     { return "Download" }
func (ds *DownloadStage) WorkerCount() int { return ds.workers }

Parte 8: Etapa de Validación — Reglas de Negocio y Cumplimiento

// domain/validation.go
package domain

type ValidationStage struct {
	validator domain.InvoiceValidator
	workers   int
	out       chan interface{}
}

func NewValidationStage(workers int) *ValidationStage {
	return &ValidationStage{
		validator: NewSATCompliantValidator(),
		workers:   workers,
		out:       make(chan interface{}, 100),
	}
}

func (vs *ValidationStage) Process(ctx context.Context, input <-chan interface{}) <-chan interface{} {
	var wg sync.WaitGroup

	for i := 0; i < vs.workers; i++ {
		wg.Add(1)
		go vs.validateWorker(ctx, input, &wg)
	}

	go func() {
		wg.Wait()
		close(vs.out)
	}()

	return vs.out
}

func (vs *ValidationStage) validateWorker(ctx context.Context, input <-chan interface{}, wg *sync.WaitGroup) {
	defer wg.Done()

	for record := range input {
		fiscal := record.(*FiscalRecord)

		// Saltar si ya falló
		if fiscal.Status != StatusDownloaded {
			vs.out <- fiscal
			continue
		}

		// Parsear XML
		invoice, err := parseInvoiceXML(fiscal.RawXMLData)
		if err != nil {
			fiscal.Status = StatusRejected
			fiscal.ErrorMessage = fmt.Sprintf("xml parse failed: %v", err)
			vs.out <- fiscal
			continue
		}

		// Validar contra reglas de dominio
		validationErrors := vs.validator.Validate(invoice)
		if len(validationErrors) > 0 {
			fiscal.Status = StatusRejected
			fiscal.ErrorMessage = fmt.Sprintf("validation failed: %v", validationErrors)
			vs.out <- fiscal
			continue
		}

		// Validación pasó
		fiscal.Status = StatusValidated
		fiscal.InvoiceData = invoice
		vs.out <- fiscal
	}
}

Parte 9: Características de Rendimiento y Benchmarks

Rendimiento Medido en Hardware de Producción

Máquina: AWS c6i.2xlarge (8 vCPU, 16GB RAM)

100,000 registros × 500ms/archivo:

Etapa	Workers	Duración	Throughput
Consulta (DB)	15	67s	1,492 registros/s
Descarga (SFTP)	25	1,000s	50 archivos/sec
Validar (XML)	20	500s (solapado)	200 archivos/sec
Comprimir (gzip)	10	300s (solapado)	333 archivos/sec
Indexar (cache)	1	50s (final)	2,000 registros/sec
Total (solapado)	—	1,200s	83 registros/sec
Serial (baseline)	—	50,000s	2 registros/sec
Mejora	—	41.7x más rápido	—

Uso de Memoria:

Por conexión SFTP: ~2MB (estado de conexión)
Por goroutine: ~2KB
50 goroutines: ~100KB
Buffer XML (50 archivos × 1MB): ~50MB
Total pipeline: <200MB
Cache (100K registros indexados): ~500MB

Parte 10: Manejo de Errores y Estrategia de Reintentos

// application/error_handler.go
package application

type ErrorHandler struct {
	maxRetries int
	backoff    BackoffStrategy
	eventBus   EventBus
}

type BackoffStrategy interface {
	NextDelay(attemptNumber int) time.Duration
}

// ExponentialBackoff: 100ms, 200ms, 400ms, 800ms, 1.6s
type ExponentialBackoff struct {
	initialDelay time.Duration
	maxDelay     time.Duration
}

func (eb *ExponentialBackoff) NextDelay(attempt int) time.Duration {
	delay := eb.initialDelay * time.Duration(math.Pow(2, float64(attempt)))
	if delay > eb.maxDelay {
		delay = eb.maxDelay
	}
	return delay
}

// Recover procesa registros fallidos e intenta reintentar
func (eh *ErrorHandler) Recover(ctx context.Context, failed []*FiscalRecord) (*RecoveryResult, error) {
	result := &RecoveryResult{
		Total:       len(failed),
		Successful:  0,
		StillFailed: 0,
	}

	for _, record := range failed {
		if record.RetryCount >= eh.maxRetries {
			result.StillFailed++

			// Publicar evento de fallo para auditoría
			eh.eventBus.Publish(&RecordFailedEvent{
				UUID:        record.UUID,
				ErrorReason: record.ErrorMessage,
				RetryCount:  record.RetryCount,
				FailedAt:    time.Now(),
			})

			continue
		}

		// Programar reintento con exponential backoff
		delay := eh.backoff.NextDelay(record.RetryCount)
		time.Sleep(delay)

		// Lógica de reintento aquí...
		if retrySuccess {
			result.Successful++
		}
	}

	return result, nil
}

Parte 11: Monitoreo y Observabilidad

// infrastructure/metrics.go
package infrastructure

type MetricsCollector interface {
	RecordStageExecution(stageName string, duration time.Duration, recordCount int)
	RecordError(stageName string, errorType string)
	RecordThroughput(recordsPerSecond float64)
}

type PrometheusMetrics struct {
	pipelineExecutionTime    prometheus.Histogram
	stageProcessingDuration  *prometheus.HistogramVec
	recordsProcessed         prometheus.Counter
	recordsFailed            prometheus.Counter
	backpressureEvents       prometheus.Counter
	cacheHitRate             prometheus.Gauge
}

// Registrar eventos clave para debugging
func (p *FiscalProcessingPipeline) LogPipelineMetrics(result *ProcessingResult) {
	log.WithFields(log.Fields{
		"job_id":     result.JobID,
		"successful": result.Successful,
		"failed":     result.Failed,
		"duration":   result.Duration,
		"throughput": float64(result.Successful) / result.Duration.Seconds(),
	}).Info("Pipeline completed")

	// Alertar si tasa de fallo es demasiado alta
	failureRate := float64(result.Failed) / float64(result.Successful+result.Failed)
	if failureRate > 0.05 { // umbral 5%
		log.WithFields(log.Fields{
			"job_id":        result.JobID,
			"failure_rate":  failureRate,
			"failed_count":  result.Failed,
		}).Warn("High failure rate detected")
	}
}

Parte 12: Recomendación de Stack de Deployment

Backend

Language: Go 1.21+
Framework: Chi (HTTP routing ligero)
Database: PostgreSQL 15 (soporte JSONB, consultas paralelas)
Cache: Redis (caching distribuido, pub/sub para eventos)
Message Queue: RabbitMQ o Kafka (distribución de trabajos)
Storage: S3 (archivos comprimidos) + NVMe local (cache caliente)

Infraestructura

graph TB
    subgraph "Load Balancing"
        LB["AWS ALB<br/>multiple zones"]
    end

    subgraph "API Servers"
        API1["Go App<br/>Container"]
        API2["Go App<br/>Container"]
        API3["Go App<br/>Container"]
    end

    subgraph "Data Layer"
        DB["PostgreSQL 15<br/>Primary + Replicas"]
        CACHE["Redis Cluster<br/>3 nodes"]
    end

    subgraph "Storage"
        S3["S3<br/>Compressed Files"]
        NVMe["EBS gp3<br/>Hot Cache"]
    end

    subgraph "External"
        SFTP["SFTP Servers<br/>Regional"]
    end

    LB -->|Round Robin| API1
    LB -->|Round Robin| API2
    LB -->|Round Robin| API3

    API1 -->|Queries| DB
    API2 -->|Queries| DB
    API3 -->|Queries| DB

    API1 -->|Cache Hits| CACHE
    API2 -->|Cache Hits| CACHE
    API3 -->|Cache Hits| CACHE

    API1 -->|Uploads| S3
    API1 -->|Hot Index| NVMe

    API1 -->|Downloads| SFTP

    DB -->|Replication| DB

    style LB fill:#2196f3
    style CACHE fill:#4caf50
    style DB fill:#ff9800

Configuración de Deployment

# docker-compose.yml ejemplo
version: '3.9'

services:
  api:
    image: fiscal-processor:latest
    ports:
      - "8080:8080"
    environment:
      DATABASE_URL: postgresql://user:pass@db:5432/fiscal
      REDIS_URL: redis://cache:6379
      SFTP_POOL_SIZE: 10
      WORKER_COUNT_QUERY: 15
      WORKER_COUNT_DOWNLOAD: 25
      WORKER_COUNT_VALIDATE: 20
      WORKER_COUNT_COMPRESS: 10
    depends_on:
      - db
      - cache

  db:
    image: postgres:15-alpine
    environment:
      POSTGRES_DB: fiscal
      POSTGRES_PASSWORD: secure_password
    volumes:
      - postgres_data:/var/lib/postgresql/data
    command:
      - "-c"
      - "max_connections=200"
      - "-c"
      - "shared_buffers=4GB"

  cache:
    image: redis:7-alpine
    command: redis-server --maxmemory 2gb --maxmemory-policy allkeys-lru

volumes:
  postgres_data:

Parte 13: La Realidad de Producción

Lo Que Esta Arquitectura Resuelve

Escalabilidad: Procesar 100,000+ registros sin fallar
Throughput: 50 archivos/segundo por servidor (1,500 archivos/minuto)
Confiabilidad: Reintento automático con exponential backoff
Observabilidad: Rastro de auditoría de eventos completo para cumplimiento
Responsividad: Respuestas API en <100ms (cacheadas), procesamiento masivo en background
Eficiencia de Costos: Mejora de 150x en throughput sobre procesamiento serial

Lo Que Aún Requiere Disciplina

Apagado elegante: Cerrar pipelines adecuadamente y descargar datos pendientes
Alertas de monitoreo: Alertar cuando tasa de fallo excede 5%
Sintonización de BD: Indexar fiscal_records por (status, issuance_date) para consultas rápidas
Límites de conexión: Servidores SFTP típicamente permiten 10-20 conexiones concurrentes máximo
Gestión de memoria: Archivos XML grandes pueden disparar memoria—hacer stream cuando sea posible

Conclusión: La Arquitectura Importa Más que la Velocidad

La diferencia entre una aplicación que procesa 100,000 registros y una que falla a 1,000 no es el lenguaje de programación. Es la arquitectura.

Worker pools. Pooling de conexiones. Manejo adecuado de errores. Event sourcing. Estas transforman caos en orden.

Go hace esto fácil porque sus primitivos de runtime (goroutines, canales, context) están diseñados exactamente para este problema.

Pero los patrones aplican a cualquier lenguaje. La disciplina es lo que importa.

El procesamiento de datos empresariales no se trata de moverse rápido. Se trata de moverse seguro a escala. Cuando diseñas para modos de fallo, ganas la confianza de procesar millones de registros con el mismo código que procesa miles. Esa confianza vale más que cualquier cantidad de optimización.