Patterns and Principles in Procedural Languages — Advanced

// => DomainEvent is the common interface every domain event must satisfy.
// => Topic() lets the bus dispatch without knowing the concrete event type.
type DomainEvent interface {
    // => Topic returns a dot-separated string: "<aggregate>.<verb>" (e.g. "po.approved")
    Topic() string
}
 
// => EventHandler is any object that can receive and process one domain event.
type EventHandler interface {
    // => Handle may return an error; the bus treats errors as fatal for that publish.
    Handle(ctx context.Context, event DomainEvent) error
}
 
// => EventBus routes domain events to registered handlers by topic.
// => Decouples producers (application services) from consumers (notification, analytics).
type EventBus struct {
    // => mu guards the handlers map for concurrent subscribe and publish safety.
    mu sync.RWMutex
    // => handlers maps each topic string to the slice of handlers subscribed to it.
    handlers map[string][]EventHandler
}
 
// => NewEventBus allocates an EventBus with an empty handler registry.
func NewEventBus() *EventBus {
    // => Make the map eagerly — nil map panics on write.
    return &EventBus{handlers: make(map[string][]EventHandler)}
}
 
// => Subscribe registers handler as a recipient of all events on topic.
func (b *EventBus) Subscribe(topic string, handler EventHandler) {
    b.mu.Lock()
    defer b.mu.Unlock()
    // => Append to existing slice — multiple handlers per topic are allowed.
    b.handlers[topic] = append(b.handlers[topic], handler)
}
 
// => Publish delivers event to every handler registered for event.Topic().
func (b *EventBus) Publish(ctx context.Context, event DomainEvent) error {
    b.mu.RLock()
    // => Read lock: concurrent publishes to different topics do not block each other.
    handlers := b.handlers[event.Topic()]
    b.mu.RUnlock()
    // => Snapshot handlers under read lock, then call them outside the lock
    // => to avoid holding the lock during potentially slow handler work.
    for _, h := range handlers {
        if err := h.Handle(ctx, event); err != nil {
            // => First handler error aborts remaining handlers for this publish.
            return fmt.Errorf("handler %T failed for topic %s: %w", h, event.Topic(), err)
        }
    }
    return nil
}
 
// --- Usage ---
 
// => POApprovedEvent is the concrete domain event emitted when a PO is approved.
type POApprovedEvent struct {
    POID       string
    SupplierID string
    Total      Money
}
 
// => Topic satisfies DomainEvent; the bus uses this string to find subscribers.
func (e POApprovedEvent) Topic() string { return "po.approved" }
 
// => Wire up: notification and analytics both subscribe to the same topic.
bus := NewEventBus()
bus.Subscribe("po.approved", notificationHandler)
bus.Subscribe("po.approved", analyticsHandler)
bus.Subscribe("invoice.received", reconcileHandler)
 
// => Publish routes the event to both notificationHandler and analyticsHandler.
err := bus.Publish(ctx, POApprovedEvent{POID: "PO-001", SupplierID: "SUP-42", Total: usd(1500)})

// => AsyncEventBus wraps a synchronous EventBus with a buffered channel
// => and a fixed-size worker pool that drains the channel concurrently.
type AsyncEventBus struct {
    // => inner is the synchronous bus; workers call inner.Publish for each event.
    inner *EventBus
    // => queue is a buffered channel; producers enqueue without blocking until full.
    queue chan asyncJob
    // => wg tracks all running workers for clean shutdown via Shutdown().
    wg sync.WaitGroup
}
 
// => asyncJob pairs an event with its context so deadline/cancellation propagates.
type asyncJob struct {
    ctx   context.Context
    event DomainEvent
}
 
// => NewAsyncEventBus creates a bus with workerCount goroutines and bufferSize slots.
func NewAsyncEventBus(inner *EventBus, workerCount, bufferSize int) *AsyncEventBus {
    b := &AsyncEventBus{
        inner: inner,
        // => Buffer absorbs bursts; choose bufferSize = peak_rps × expected_handler_ms / 1000.
        queue: make(chan asyncJob, bufferSize),
    }
    b.wg.Add(workerCount)
    for i := 0; i < workerCount; i++ {
        go func() {
            defer b.wg.Done()
            // => Each worker loops until the queue channel is closed.
            for job := range b.queue {
                // => Errors are logged here; callers never see handler errors.
                if err := b.inner.Publish(job.ctx, job.event); err != nil {
                    log.Printf("async handler error: %v", err)
                }
            }
        }()
    }
    return b
}
 
// => Enqueue drops the event into the channel; returns immediately.
// => Returns ErrQueueFull if the buffer is exhausted — caller handles backpressure.
var ErrQueueFull = errors.New("async event queue is full")
 
func (b *AsyncEventBus) Enqueue(ctx context.Context, event DomainEvent) error {
    select {
    case b.queue <- asyncJob{ctx: ctx, event: event}:
        // => Happy path: event accepted without blocking.
        return nil
    default:
        // => Default branch fires immediately when channel is full (non-blocking).
        return ErrQueueFull
    }
}
 
// => Shutdown closes the channel and waits for all workers to drain the queue.
func (b *AsyncEventBus) Shutdown() {
    // => Close signals workers to stop after draining remaining jobs.
    close(b.queue)
    b.wg.Wait()
}

// => POApprovalService uses an AsyncEventBus and must handle ErrQueueFull.
type POApprovalService struct {
    repo     PORepository
    eventBus *AsyncEventBus
    logger   Logger
}
 
// => ApprovePO executes the approval business logic and enqueues the result event.
func (s *POApprovalService) ApprovePO(ctx context.Context, poID string) error {
    po, err := s.repo.FindByID(ctx, poID)
    if err != nil {
        return fmt.Errorf("load PO: %w", err)
    }
    // => Approve mutates status to Approved and returns the event to emit.
    event, err := po.Approve()
    if err != nil {
        return fmt.Errorf("approve: %w", err)
    }
    if err := s.repo.Save(ctx, po); err != nil {
        return fmt.Errorf("save: %w", err)
    }
    // => Enqueue the event; the domain state is already persisted above.
    if err := s.eventBus.Enqueue(ctx, event); err != nil {
        // => Queue full: log and continue — state was saved; event delivery is best-effort.
        // => In high-assurance systems, persist the event in an outbox table here instead.
        s.logger.Error("event queue full, dropping POApproved event",
            err,
            F("po_id", poID))
        // => Return nil: the business operation succeeded even if event delivery is delayed.
        return nil
    }
    return nil
}
 
// --- HTTP layer backpressure translation ---
 
// => HandleApprove translates ErrQueueFull into HTTP 429 Too Many Requests.
func HandleApprove(svc *POApprovalService) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        poID := r.PathValue("id")
        if err := svc.ApprovePO(r.Context(), poID); err != nil {
            // => Distinguish transient overload (503/429) from permanent failure (400/500).
            if errors.Is(err, ErrQueueFull) {
                w.Header().Set("Retry-After", "5")
                http.Error(w, "server overloaded", http.StatusTooManyRequests)
                return
            }
            http.Error(w, err.Error(), http.StatusInternalServerError)
            return
        }
        w.WriteHeader(http.StatusNoContent)
    }
}

// => EventStore is the append-only log for domain events keyed by aggregate ID.
// => No update or delete operations exist — immutability is the guarantee.
type EventStore interface {
    // => Append adds events to the end of the stream for aggregateID.
    // => expectedVersion enables optimistic concurrency: fail if stream has grown.
    Append(ctx context.Context, aggregateID string, expectedVersion int, events []DomainEvent) error
    // => Load returns all events for aggregateID in the order they were appended.
    Load(ctx context.Context, aggregateID string) ([]DomainEvent, error)
}
 
// => InMemoryEventStore is a test double; production uses PostgreSQL or EventStoreDB.
type InMemoryEventStore struct {
    mu     sync.RWMutex
    // => streams maps aggregateID to its ordered event slice.
    streams map[string][]DomainEvent
}
 
func (s *InMemoryEventStore) Append(
    ctx context.Context, aggregateID string, expectedVersion int, events []DomainEvent,
) error {
    s.mu.Lock()
    defer s.mu.Unlock()
    existing := s.streams[aggregateID]
    // => Optimistic concurrency check: reject if another writer has appended since last read.
    if len(existing) != expectedVersion {
        return fmt.Errorf("concurrency conflict: expected version %d, got %d",
            expectedVersion, len(existing))
    }
    s.streams[aggregateID] = append(existing, events...)
    return nil
}
 
func (s *InMemoryEventStore) Load(
    ctx context.Context, aggregateID string,
) ([]DomainEvent, error) {
    s.mu.RLock()
    defer s.mu.RUnlock()
    // => Return a copy to prevent callers from mutating the stored slice.
    events := make([]DomainEvent, len(s.streams[aggregateID]))
    copy(events, s.streams[aggregateID])
    return events, nil
}
 
// => RehydratePO reconstructs a PurchaseOrder from its full event history.
func RehydratePO(ctx context.Context, store EventStore, poID string) (*PurchaseOrder, error) {
    events, err := store.Load(ctx, poID)
    if err != nil {
        return nil, fmt.Errorf("load events: %w", err)
    }
    po := &PurchaseOrder{} // => Start from zero value — no current-state row needed.
    for _, event := range events {
        // => Apply each event in sequence to build up the current state.
        if err := po.Apply(event); err != nil {
            return nil, fmt.Errorf("apply %T: %w", event, err)
        }
    }
    // => po now reflects the aggregate state as of its latest event.
    return po, nil
}

// => POSummary is the read-model DTO optimized for list-page queries.
// => It contains denormalized data so the query needs no joins.
type POSummary struct {
    ID           string
    Status       POStatus
    SupplierName string
    TotalAmount  Money
    // => UpdatedAt tracks when the projection last processed an event for this PO.
    UpdatedAt time.Time
}
 
// => POSummaryProjection maintains the in-memory (or DB-backed) read model.
type POSummaryProjection struct {
    mu      sync.RWMutex
    // => summaries maps PO ID to its current denormalized summary.
    summaries map[string]*POSummary
}
 
// => Handle routes incoming domain events to the appropriate update method.
func (p *POSummaryProjection) Handle(ctx context.Context, event DomainEvent) error {
    switch e := event.(type) {
    case POCreatedEvent:
        // => POCreated initializes the summary row with Draft status.
        return p.onPOCreated(e)
    case POApprovedEvent:
        // => POApproved updates status without touching immutable fields.
        return p.onPOApproved(e)
    case POCancelledEvent:
        // => POCancelled marks the summary as cancelled; row is retained for audit.
        return p.onPOCancelled(e)
    default:
        // => Unknown events are silently ignored — projections are additive.
        return nil
    }
}
 
func (p *POSummaryProjection) onPOCreated(e POCreatedEvent) error {
    p.mu.Lock()
    defer p.mu.Unlock()
    p.summaries[e.POID] = &POSummary{
        ID:           e.POID,
        Status:       POStatusDraft,
        SupplierName: e.SupplierName,
        TotalAmount:  e.Total,
        UpdatedAt:    e.OccurredAt,
    }
    return nil
}
 
func (p *POSummaryProjection) onPOApproved(e POApprovedEvent) error {
    p.mu.Lock()
    defer p.mu.Unlock()
    s, ok := p.summaries[e.POID]
    if !ok {
        // => Out-of-order event: the create event may not have arrived yet.
        // => Production: store in a pending queue and replay after create arrives.
        return fmt.Errorf("projection gap: PO %s not found for POApproved", e.POID)
    }
    s.Status = POStatusApproved
    s.UpdatedAt = e.OccurredAt
    return nil
}
 
// => GetSummaries returns all current PO summaries under a read lock.
func (p *POSummaryProjection) GetSummaries() []POSummary {
    p.mu.RLock()
    defer p.mu.RUnlock()
    result := make([]POSummary, 0, len(p.summaries))
    for _, s := range p.summaries {
        result = append(result, *s)
    }
    return result
}

// => Logger interface: all application code depends on this abstraction, not on
// => a specific logging library (zerolog, zap, slog). Swap implementations freely.
type Logger interface {
    // => Info logs a message at INFO level with structured key-value fields.
    Info(msg string, fields ...Field)
    // => Error logs at ERROR level; error is a first-class parameter, not just a field,
    // => so log management systems can index it specially.
    Error(msg string, err error, fields ...Field)
    // => With returns a child logger pre-populated with the given fields.
    // => Use With to attach request-scoped data (po_id, user_id) once at the boundary.
    With(fields ...Field) Logger
}
 
// => Field is a typed key-value pair for log entries.
// => Using a struct (not map[string]interface{}) keeps allocations predictable.
type Field struct {
    Key   string
    Value interface{}
}
 
// => F is a shorthand constructor; call sites read as English: F("po_id", po.ID).
func F(key string, value interface{}) Field {
    return Field{Key: key, Value: value}
}
 
// --- Usage in application service ---
 
func (s *POApprovalService) ApprovePO(ctx context.Context, poID string) error {
    start := time.Now()
    // => Attach po_id once; all subsequent log calls on log inherit it.
    log := s.logger.With(F("po_id", poID))
 
    po, err := s.repo.FindByID(ctx, poID)
    if err != nil {
        log.Error("failed to load purchase order", err)
        return fmt.Errorf("load PO: %w", err)
    }
 
    event, err := po.Approve()
    if err != nil {
        // => Domain error: log at INFO level — not an infrastructure failure.
        log.Info("purchase_order_approval_rejected",
            F("reason", err.Error()),
            F("po_status", po.Status.String()))
        return err
    }
 
    if err := s.repo.Save(ctx, po); err != nil {
        log.Error("failed to save approved purchase order", err)
        return fmt.Errorf("save: %w", err)
    }
 
    // => Successful approval: log all fields needed for business analytics.
    log.Info("purchase_order_approved",
        F("supplier_id", po.SupplierID),
        F("total", po.Total.String()),
        F("duration_ms", time.Since(start).Milliseconds()))
    return nil
}

// => correlationIDKey is an unexported type to prevent key collisions in context.
// => Using a custom type (not string) ensures only this package can read the value.
type contextKey string
 
const correlationIDKey contextKey = "correlation_id"
 
// => CorrelationIDMiddleware reads X-Correlation-ID from the request or generates one.
func CorrelationIDMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        // => Accept existing correlation ID from upstream service (distributed trace).
        correlationID := r.Header.Get("X-Correlation-ID")
        if correlationID == "" {
            // => Generate a new ID when this service is the entry point.
            correlationID = uuid.New().String()
        }
        // => Echo the correlation ID back so callers can correlate their own logs.
        w.Header().Set("X-Correlation-ID", correlationID)
        // => Inject into context so all downstream calls in this request can read it.
        ctx := context.WithValue(r.Context(), correlationIDKey, correlationID)
        next.ServeHTTP(w, r.WithContext(ctx))
    })
}
 
// => CorrelationIDFromContext extracts the ID; returns empty string if absent.
func CorrelationIDFromContext(ctx context.Context) string {
    id, _ := ctx.Value(correlationIDKey).(string)
    return id
}
 
// => LoggerFromContext returns a logger pre-populated with the correlation ID.
// => Call this once per handler to get a request-scoped logger.
func LoggerFromContext(ctx context.Context, base Logger) Logger {
    id := CorrelationIDFromContext(ctx)
    if id == "" {
        return base
    }
    // => All log entries from this request will carry correlation_id automatically.
    return base.With(F("correlation_id", id))
}
 
// --- Usage in HTTP handler ---
 
func HandleApprovePO(svc *POApprovalService, baseLogger Logger) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        // => One call; all log entries in the request lifecycle carry the ID.
        log := LoggerFromContext(r.Context(), baseLogger)
        poID := r.PathValue("id")
        if err := svc.ApprovePO(r.Context(), poID); err != nil {
            log.Error("approve_po failed", err, F("po_id", poID))
            http.Error(w, "internal error", http.StatusInternalServerError)
            return
        }
        w.WriteHeader(http.StatusNoContent)
    }
}

// => MetricsCollector abstracts the metrics backend (Prometheus, StatsD, CloudWatch).
// => Application code calls this interface; the binding is configured at startup.
type MetricsCollector interface {
    // => IncrCounter increments a named counter by 1.
    // => labels is a flat key-value slice: ["po_status", "approved", "supplier_tier", "gold"].
    IncrCounter(name string, labels map[string]string)
    // => RecordHistogram records one observation of a continuous value.
    // => value unit depends on the metric name (milliseconds, bytes, items).
    RecordHistogram(name string, value float64, labels map[string]string)
}
 
// => PrometheusMetrics implements MetricsCollector backed by the Prometheus client.
type PrometheusMetrics struct {
    // => counters and histograms are registered at startup; never registered on hot path.
    counters   map[string]*prometheus.CounterVec
    histograms map[string]*prometheus.HistogramVec
}
 
func (m *PrometheusMetrics) IncrCounter(name string, labels map[string]string) {
    c, ok := m.counters[name]
    if !ok {
        return // => Unknown metric: silently skip rather than panic.
    }
    // => With() resolves the label set; Inc() is atomic — safe for concurrent calls.
    c.With(prometheus.Labels(labels)).Inc()
}
 
func (m *PrometheusMetrics) RecordHistogram(name string, value float64, labels map[string]string) {
    h, ok := m.histograms[name]
    if !ok {
        return
    }
    h.With(prometheus.Labels(labels)).Observe(value)
}
 
// --- Usage: instrument POApprovalService ---
 
func (s *POApprovalService) ApprovePO(ctx context.Context, poID string) error {
    start := time.Now()
    _, err := s.approveCore(ctx, poID)
    elapsed := time.Since(start).Seconds() * 1000 // => Convert to milliseconds.
 
    status := "success"
    if err != nil {
        status = "failure"
    }
    // => Counter: total approvals segmented by outcome.
    s.metrics.IncrCounter("po_approvals_total", map[string]string{"status": status})
    // => Histogram: latency distribution for SLA alerting (p99 > 500ms = alert).
    s.metrics.RecordHistogram("po_approval_duration_ms",
        elapsed,
        map[string]string{"status": status})
    return err
}

// => Tracer interface: application code creates spans without importing otel directly.
// => This keeps the domain and application layers free of telemetry SDK dependencies.
type Tracer interface {
    // => StartSpan creates a child span under any span already in ctx.
    // => Returns a new context carrying the child span and the span itself.
    StartSpan(ctx context.Context, name string) (context.Context, Span)
}
 
// => Span represents one unit of traced work.
type Span interface {
    // => End records the span's end time; always call in a defer.
    End()
    // => SetAttribute attaches a key-value annotation to the span.
    SetAttribute(key string, value interface{})
    // => RecordError marks the span as failed and records the error message.
    RecordError(err error)
}
 
// --- Usage in HTTP handler and repository ---
 
// => HandleApprovePO creates the root span for the HTTP request.
func HandleApprovePO(tracer Tracer, svc *POApprovalService) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        // => Root span: name matches the operation visible in the trace UI.
        ctx, span := tracer.StartSpan(r.Context(), "http.POST /purchase-orders/:id/approve")
        defer span.End()
        // => Attach HTTP metadata so the span is searchable by route.
        span.SetAttribute("http.method", "POST")
        span.SetAttribute("http.route", "/purchase-orders/:id/approve")
 
        poID := r.PathValue("id")
        if err := svc.ApprovePO(ctx, poID); err != nil {
            // => Record error on the span; trace backend shows failed spans in red.
            span.RecordError(err)
            http.Error(w, "internal error", http.StatusInternalServerError)
            return
        }
        w.WriteHeader(http.StatusNoContent)
    }
}
 
// => PostgresPORepository creates child spans so DB latency is visible separately.
func (r *PostgresPORepository) FindByID(ctx context.Context, id string) (*PurchaseOrder, error) {
    // => Child span: inherits parent trace ID from ctx automatically.
    ctx, span := r.tracer.StartSpan(ctx, "db.purchase_orders.find_by_id")
    defer span.End()
    span.SetAttribute("db.statement", "SELECT * FROM purchase_orders WHERE id = $1")
    span.SetAttribute("po_id", id)
    // => Execute query; span.End() records when the query returned.
    return r.queryPO(ctx, id)
}

// => HealthStatus represents the result of one dependency health check.
type HealthStatus struct {
    // => OK is false when this dependency is unavailable or degraded.
    OK      bool
    // => Message carries a human-readable description for the health endpoint response.
    Message string
}
 
// => HealthChecker can check the health of one dependency (DB, cache, vendor API).
type HealthChecker interface {
    // => Name identifies this check in the composite response JSON.
    Name() string
    // => Check performs the health probe; ctx carries a short deadline (e.g. 2 seconds).
    Check(ctx context.Context) HealthStatus
}
 
// => CompositeHealthChecker aggregates multiple HealthCheckers into one status.
type CompositeHealthChecker struct {
    checkers []HealthChecker
}
 
// => Check returns aggregated status: OK only if all constituent checkers pass.
func (c *CompositeHealthChecker) Check(ctx context.Context) map[string]HealthStatus {
    results := make(map[string]HealthStatus, len(c.checkers))
    for _, checker := range c.checkers {
        // => Run each check with the shared deadline; one slow check blocks others.
        results[checker.Name()] = checker.Check(ctx)
    }
    return results
}
 
// => HealthHandler exposes GET /health returning JSON status per dependency.
func HealthHandler(composite *CompositeHealthChecker) http.HandlerFunc {
    return func(w http.ResponseWriter, r *http.Request) {
        // => Short deadline: health checks must respond faster than the probe timeout.
        ctx, cancel := context.WithTimeout(r.Context(), 2*time.Second)
        defer cancel()
 
        checks := composite.Check(ctx)
 
        // => Determine overall status: any failing check makes the whole response unhealthy.
        allOK := true
        for _, s := range checks {
            if !s.OK {
                allOK = false
                break
            }
        }
 
        w.Header().Set("Content-Type", "application/json")
        if !allOK {
            // => 503 signals to Kubernetes readiness probe that this pod should be drained.
            w.WriteHeader(http.StatusServiceUnavailable)
        }
        // => Encode checks map as JSON: {"db":{"ok":true},"cache":{"ok":false,"message":"..."}}
        json.NewEncoder(w).Encode(map[string]interface{}{
            "status": map[bool]string{true: "ok", false: "degraded"}[allOK],
            "checks": checks,
        })
    }
}
 
// --- Example checker implementations ---
 
// => DBHealthChecker verifies the database connection with a lightweight ping.
type DBHealthChecker struct{ db *sql.DB }
 
func (c *DBHealthChecker) Name() string { return "db" }
func (c *DBHealthChecker) Check(ctx context.Context) HealthStatus {
    // => PingContext uses the context deadline — returns error if DB is unreachable.
    if err := c.db.PingContext(ctx); err != nil {
        return HealthStatus{OK: false, Message: err.Error()}
    }
    return HealthStatus{OK: true}
}

// => SagaOrchestrator drives a long-running business process as an explicit state machine.
// => Re-entrant: safe to call Execute again if the previous call was interrupted mid-saga.
type SagaOrchestrator interface {
    // => Execute runs the saga from its current persisted state to completion or failure.
    Execute(ctx context.Context, sagaID SagaID) error
}
 
// => SagaStep pairs one unit of forward work with its compensating transaction.
type SagaStep struct {
    // => Name is the human-readable step identifier used for logging and state persistence.
    Name string
    // => Action performs the forward operation; must be idempotent (see Example 70).
    Action func(ctx context.Context) error
    // => Compensation undoes Action if a later step fails; also must be idempotent.
    Compensation func(ctx context.Context) error
}
 
// => SagaID is a typed alias to prevent passing a plain string in the wrong parameter.
type SagaID string
 
// --- Procurement saga steps ---
 
// => buildPOApprovalSteps constructs the ordered steps for PO approval.
func buildPOApprovalSteps(po *PurchaseOrder, svc *POApprovalDeps) []SagaStep {
    return []SagaStep{
        {
            Name: "reserve_supplier_credit",
            // => Reserve credit before approval — prevents over-commitment.
            Action:       func(ctx context.Context) error { return svc.creditSvc.Reserve(ctx, po) },
            Compensation: func(ctx context.Context) error { return svc.creditSvc.Release(ctx, po) },
        },
        {
            Name: "notify_l1_manager",
            Action:       func(ctx context.Context) error { return svc.notifier.NotifyL1(ctx, po) },
            // => Compensation sends a cancellation notification so the manager is not left waiting.
            Compensation: func(ctx context.Context) error { return svc.notifier.CancelL1(ctx, po) },
        },
        {
            Name: "update_po_status",
            Action:       func(ctx context.Context) error { return svc.repo.SetStatus(ctx, po.ID, POStatusPendingApproval) },
            // => Revert to Draft if the saga cannot complete.
            Compensation: func(ctx context.Context) error { return svc.repo.SetStatus(ctx, po.ID, POStatusDraft) },
        },
    }
}

// => runSteps executes steps in order; compensates in reverse on the first failure.
func runSteps(ctx context.Context, steps []SagaStep) error {
    // => completed tracks how many steps succeeded so we compensate exactly those.
    completed := make([]SagaStep, 0, len(steps))
 
    for _, step := range steps {
        if err := step.Action(ctx); err != nil {
            // => This step failed: compensate all previously completed steps in LIFO order.
            compensationErr := compensate(ctx, completed)
            if compensationErr != nil {
                // => Compensation also failed — log and surface both errors.
                return fmt.Errorf("saga failed at %s: %w; compensation error: %v",
                    step.Name, err, compensationErr)
            }
            return fmt.Errorf("saga failed at step %s: %w", step.Name, err)
        }
        // => Step succeeded: record it so we can compensate it if a later step fails.
        completed = append(completed, step)
    }
    // => All steps succeeded — saga complete.
    return nil
}
 
// => compensate runs all completed steps' compensations in reverse (LIFO) order.
func compensate(ctx context.Context, completed []SagaStep) error {
    // => Iterate backward: last completed step is compensated first.
    for i := len(completed) - 1; i >= 0; i-- {
        step := completed[i]
        if err := step.Compensation(ctx); err != nil {
            // => Compensation failure leaves the system in a partially-compensated state.
            // => Production: persist to a dead-letter table for manual intervention.
            return fmt.Errorf("compensation of step %s failed: %w", step.Name, err)
        }
    }
    return nil
}

// => SagaStateStore persists which steps have already completed for a given saga.
// => This is the idempotency journal; backed by a database in production.
type SagaStateStore interface {
    // => IsStepCompleted returns true if stepName was previously marked done for sagaID.
    IsStepCompleted(ctx context.Context, sagaID SagaID, stepName string) (bool, error)
    // => MarkStepCompleted records that stepName completed successfully for sagaID.
    MarkStepCompleted(ctx context.Context, sagaID SagaID, stepName string) error
}
 
// => idempotentAction wraps any SagaStep action with an idempotency check.
// => If the step was already completed, it is a no-op on re-run.
func idempotentAction(
    store SagaStateStore,
    sagaID SagaID,
    step SagaStep,
) func(ctx context.Context) error {
    return func(ctx context.Context) error {
        done, err := store.IsStepCompleted(ctx, sagaID, step.Name)
        if err != nil {
            return fmt.Errorf("check idempotency for %s: %w", step.Name, err)
        }
        if done {
            // => Step already ran — skip to preserve exactly-once semantics.
            return nil
        }
        // => Execute the real action.
        if err := step.Action(ctx); err != nil {
            return err
        }
        // => Mark completed only after success — if MarkStepCompleted fails here,
        // => the step will re-run on retry, but it must also be idempotent itself.
        return store.MarkStepCompleted(ctx, sagaID, step.Name)
    }
}
 
// => wrapStepsWithIdempotency returns a new slice of steps with idempotency guards.
func wrapStepsWithIdempotency(store SagaStateStore, sagaID SagaID, steps []SagaStep) []SagaStep {
    wrapped := make([]SagaStep, len(steps))
    for i, step := range steps {
        wrapped[i] = SagaStep{
            Name:         step.Name,
            Action:       idempotentAction(store, sagaID, step),
            // => Compensations are NOT wrapped — always run compensation even if
            // => the step was previously marked completed (conservative approach).
            Compensation: step.Compensation,
        }
    }
    return wrapped
}

// => In choreography, ProcurementService emits POCreated and moves on.
// => It has no knowledge of what happens next — no coupling to ApprovalService.
func (s *ProcurementService) SubmitPO(ctx context.Context, cmd SubmitPOCommand) error {
    po, err := NewPurchaseOrder(cmd)
    if err != nil {
        return fmt.Errorf("create PO: %w", err)
    }
    if err := s.repo.Save(ctx, po); err != nil {
        return fmt.Errorf("save PO: %w", err)
    }
    // => Emit the domain event; downstream services react independently.
    return s.bus.Publish(ctx, POCreatedEvent{POID: po.ID, SupplierID: po.SupplierID, Total: po.Total})
}
 
// => ApprovalService subscribes to POCreated and emits POApproved when done.
// => It is unaware of who published POCreated or who will consume POApproved.
type ApprovalService struct {
    repo   PORepository
    bus    *EventBus
    policy ApprovalPolicy
}
 
func (s *ApprovalService) Handle(ctx context.Context, event DomainEvent) error {
    e, ok := event.(POCreatedEvent)
    if !ok {
        return nil // => Ignore events this service does not own.
    }
    po, err := s.repo.FindByID(ctx, e.POID)
    if err != nil {
        return fmt.Errorf("load PO: %w", err)
    }
    decision := s.policy.Evaluate(po)
    if decision.Approved {
        // => Publish POApproved — NotificationService will react to this.
        return s.bus.Publish(ctx, POApprovedEvent{POID: po.ID, SupplierID: po.SupplierID, Total: po.Total})
    }
    // => Rejected: emit PORejected; no orchestrator needed to route this outcome.
    return s.bus.Publish(ctx, PORejectedEvent{POID: po.ID, Reason: decision.Reason})
}
 
// --- Trade-off commentary ---
// => Choreography advantage: services have no compile-time dependency on each other.
// => Choreography disadvantage: the overall workflow exists only as emergent behavior
// => across N event handlers — there is no single place to read the business process.
// => Debugging requires correlating events across multiple services' logs.

// => SagaError classifies failures so the execution engine can choose the right strategy.
type SagaError struct {
    // => Retryable is true for transient failures (network timeout, temporary unavailability).
    Retryable bool
    // => Cause is the underlying error for logging and wrapping.
    Cause     error
    // => Step identifies which step produced this error.
    Step      string
}
 
func (e *SagaError) Error() string {
    return fmt.Sprintf("saga step %s failed (retryable=%v): %v", e.Step, e.Retryable, e.Cause)
}
 
func (e *SagaError) Unwrap() error { return e.Cause }
 
// => runWithRetry retries an action up to maxAttempts times for retryable errors only.
func runWithRetry(ctx context.Context, step SagaStep, maxAttempts int) error {
    var lastErr error
    for attempt := 1; attempt <= maxAttempts; attempt++ {
        err := step.Action(ctx)
        if err == nil {
            return nil // => Success.
        }
        var sagaErr *SagaError
        if errors.As(err, &sagaErr) && sagaErr.Retryable {
            // => Transient: back off and retry.
            lastErr = err
            // => Exponential back-off omitted for brevity; use time.Sleep with jitter.
            continue
        }
        // => Permanent failure: do not retry, propagate immediately for compensation.
        return err
    }
    // => Exhausted retries: treat as permanent so saga compensates.
    return fmt.Errorf("step %s failed after %d attempts: %w", step.Name, maxAttempts, lastErr)
}
 
// --- Concrete classification examples ---
// => Network timeout → Retryable: true  (infrastructure transient)
// => Supplier credit limit exceeded → Retryable: false  (business rule violation)
// => Database deadlock → Retryable: true  (transient concurrency conflict)
// => PO already approved → Retryable: false  (permanent state conflict)

// => main() is the composition root: all adapters, services, and handlers are wired here.
// => Nothing is constructed anywhere else — the full dependency graph lives in one function.
func main() {
    // --- Infrastructure adapters ---
    // => DB connection established first; all repositories depend on it.
    db, err := sql.Open("postgres", os.Getenv("DATABASE_URL"))
    if err != nil {
        log.Fatalf("open db: %v", err)
    }
 
    // => Logger constructed before services so it can be injected everywhere.
    logger := newZerologLogger(os.Stdout)
    metrics := newPrometheusMetrics()
    tracer := newOtelTracer("procurement-api")
    eventBus := NewEventBus()
 
    // --- Repositories (secondary adapters) ---
    poRepo := postgres.NewPORepository(db, tracer)
    supplierRepo := postgres.NewSupplierRepository(db, tracer)
    invoiceRepo := postgres.NewInvoiceRepository(db, tracer)
 
    // --- Domain services ---
    approvalPolicy := domain.NewThresholdApprovalPolicy(
        // => Policy loaded from environment to enable per-environment configuration.
        money.MustParse(os.Getenv("L1_THRESHOLD_USD")),
        money.MustParse(os.Getenv("L2_THRESHOLD_USD")),
    )
 
    // --- Application services (use-cases) ---
    // => Each service receives exactly what it needs — no service receives the DB directly.
    poApprovalSvc := application.NewPOApprovalService(poRepo, approvalPolicy, eventBus, logger, metrics)
    invoiceSvc := application.NewInvoiceService(invoiceRepo, poRepo, logger, metrics)
 
    // --- Event handlers (subscriptions wired here, not inside services) ---
    // => Subscriptions are explicit: readable at a glance from the composition root.
    notificationHandler := notification.NewPOApprovedHandler(logger)
    eventBus.Subscribe("po.approved", notificationHandler)
    eventBus.Subscribe("invoice.received", notification.NewInvoiceReceivedHandler(logger))
 
    // --- HTTP handlers (primary adapters) ---
    mux := http.NewServeMux()
    mux.HandleFunc("POST /purchase-orders/{id}/approve", HandleApprovePO(poApprovalSvc, logger, tracer))
    mux.HandleFunc("POST /invoices", HandleCreateInvoice(invoiceSvc, logger, tracer))
    mux.HandleFunc("GET /health", HealthHandler(newCompositeChecker(db)))
 
    // => Single listen call; all wiring is complete above this line.
    log.Fatal(http.ListenAndServe(":8080", mux))
}

// --- Functional core: pure, no side effects, fully testable ---
 
// => Decision captures the outcome of the approval policy evaluation.
type Decision struct {
    // => Approved is the primary outcome.
    Approved bool
    // => RequiredLevel indicates which manager tier must sign off (1, 2, or 3).
    RequiredLevel ApprovalLevel
    // => Events holds the domain events this decision produces — not yet published.
    Events []DomainEvent
}
 
// => Decide is a pure function: same inputs always produce same outputs.
// => No DB, no HTTP, no clock — completely deterministic and fast to test.
func Decide(po PurchaseOrder, policy ApprovalPolicy) (Decision, error) {
    if po.Status != POStatusDraft {
        // => Business rule: only draft POs can be submitted for approval.
        return Decision{}, fmt.Errorf("PO %s is not in Draft status", po.ID)
    }
    level := policy.RequiredLevel(po.Total)
    // => Construct the approval event in the pure function; the shell will publish it.
    events := []DomainEvent{
        POSubmittedForApprovalEvent{
            POID:          po.ID,
            SupplierID:    po.SupplierID,
            RequiredLevel: level,
        },
    }
    return Decision{Approved: false, RequiredLevel: level, Events: events}, nil
}
 
// --- Imperative shell: reads from DB, calls pure core, writes side effects ---
 
// => POApprovalShell orchestrates the impure operations around the pure core.
type POApprovalShell struct {
    repo     PORepository
    eventBus *EventBus
    logger   Logger
}
 
func (s *POApprovalShell) SubmitForApproval(ctx context.Context, poID string) error {
    // => Impure: load from DB.
    po, err := s.repo.FindByID(ctx, poID)
    if err != nil {
        return fmt.Errorf("load: %w", err)
    }
    // => Pure: all decision logic; testable without any infrastructure.
    decision, err := Decide(*po, s.getPolicy())
    if err != nil {
        return err
    }
    po.RequiredApprovalLevel = decision.RequiredLevel
    po.Status = POStatusPendingApproval
    // => Impure: persist the updated state.
    if err := s.repo.Save(ctx, po); err != nil {
        return fmt.Errorf("save: %w", err)
    }
    // => Impure: publish all events produced by the pure core.
    for _, event := range decision.Events {
        if err := s.eventBus.Publish(ctx, event); err != nil {
            s.logger.Error("failed to publish event", err, F("event", fmt.Sprintf("%T", event)))
        }
    }
    return nil
}

// => Go does not have parameterized types on structs in the same way Rust does,
// => but distinct types per stage achieve the same compile-time safety.
 
// => DraftPO is a PO in the Draft stage. Only Submit() is callable on it.
type DraftPO struct {
    ID         string
    SupplierID string
    LineItems  []LineItem
    Total      Money
}
 
// => SubmittedPO is a PO after submission. Only Approve() or Reject() are callable.
type SubmittedPO struct {
    ID            string
    SupplierID    string
    LineItems     []LineItem
    Total         Money
    // => SubmittedAt is only present after the Submit transition — not in DraftPO.
    SubmittedAt   time.Time
    RequiredLevel ApprovalLevel
}
 
// => ApprovedPO is a PO after approval. Only CreateGRN() is callable.
type ApprovedPO struct {
    ID           string
    SupplierID   string
    Total        Money
    ApprovedAt   time.Time
    ApprovedByID string
}
 
// => Submit transitions DraftPO → SubmittedPO.
// => The return type makes it impossible to skip the Draft stage.
func Submit(draft DraftPO, policy ApprovalPolicy) (SubmittedPO, error) {
    if len(draft.LineItems) == 0 {
        return SubmittedPO{}, errors.New("cannot submit PO with no line items")
    }
    return SubmittedPO{
        ID:            draft.ID,
        SupplierID:    draft.SupplierID,
        LineItems:     draft.LineItems,
        Total:         draft.Total,
        SubmittedAt:   time.Now(),
        RequiredLevel: policy.RequiredLevel(draft.Total),
    }, nil
}
 
// => Approve transitions SubmittedPO → ApprovedPO.
// => Cannot be called on a DraftPO — different type, different function signature.
func Approve(submitted SubmittedPO, approverID string) (ApprovedPO, error) {
    if approverID == "" {
        return ApprovedPO{}, errors.New("approver ID required")
    }
    return ApprovedPO{
        ID:           submitted.ID,
        SupplierID:   submitted.SupplierID,
        Total:        submitted.Total,
        ApprovedAt:   time.Now(),
        ApprovedByID: approverID,
    }, nil
}

// => ApprovalPolicy encodes the approval thresholds and required level for one supplier tier.
type ApprovalPolicy struct {
    // => L1Threshold: POs up to this amount need only L1 manager approval.
    L1Threshold Money
    // => L2Threshold: POs between L1Threshold and L2Threshold need L2 (senior manager).
    L2Threshold Money
    // => Above L2Threshold: requires L3 (CFO) approval.
}
 
// => RequiredLevel evaluates total against the policy thresholds.
func (p ApprovalPolicy) RequiredLevel(total Money) ApprovalLevel {
    switch {
    case total.LessThanOrEqual(p.L1Threshold):
        return ApprovalLevelL1
    case total.LessThanOrEqual(p.L2Threshold):
        return ApprovalLevelL2
    default:
        return ApprovalLevelL3
    }
}
 
// => PolicyTable maps supplier tier names to their configured ApprovalPolicy.
// => Adding a new tier (e.g. "preferred") requires only a new map entry, not code changes.
type PolicyTable struct {
    // => policies is the configuration table; keyed by supplier tier string.
    policies map[string]ApprovalPolicy
    // => defaultPolicy is used when the supplier tier is unrecognized.
    defaultPolicy ApprovalPolicy
}
 
// => Lookup returns the policy for supplierTier, or the default if not found.
func (t *PolicyTable) Lookup(supplierTier string) ApprovalPolicy {
    if p, ok := t.policies[supplierTier]; ok {
        return p
    }
    // => Unknown tier falls back to the most restrictive (default) policy.
    return t.defaultPolicy
}
 
// --- Table populated at startup from configuration ---
var defaultPolicyTable = &PolicyTable{
    policies: map[string]ApprovalPolicy{
        // => Gold-tier suppliers have higher thresholds (trusted, long relationship).
        "gold": {L1Threshold: usd(10000), L2Threshold: usd(50000)},
        // => Silver-tier suppliers have standard thresholds.
        "silver": {L1Threshold: usd(5000), L2Threshold: usd(25000)},
        // => Bronze-tier suppliers have the lowest thresholds (new or unvetted).
        "bronze": {L1Threshold: usd(1000), L2Threshold: usd(5000)},
    },
    // => Default: treat unknown tiers as bronze (conservative).
    defaultPolicy: ApprovalPolicy{L1Threshold: usd(1000), L2Threshold: usd(5000)},
}

// --- Test double factories ---
 
// => TestPOFactory creates pre-configured PurchaseOrder values for common test scenarios.
// => Each constructor captures the intent in its name: no need to read field assignments.
type TestPOFactory struct{}
 
// => DraftPO returns a minimal valid PO in Draft status for basic operation tests.
func (TestPOFactory) DraftPO() DraftPO {
    return DraftPO{
        ID:         "PO-TEST-001",
        SupplierID: "SUP-TEST-001",
        // => Single line item makes total predictable without reading all fields.
        LineItems: []LineItem{{ProductID: "PROD-001", Quantity: qty(10), UnitPrice: usd(100)}},
        Total:     usd(1000),
    }
}
 
// => HighValuePO returns a PO whose total exceeds L2 approval threshold.
// => Tests for CFO approval path use this without hardcoding magic numbers.
func (TestPOFactory) HighValuePO() DraftPO {
    return DraftPO{
        ID:         "PO-TEST-HIGHVAL",
        SupplierID: "SUP-TEST-001",
        LineItems:  []LineItem{{ProductID: "PROD-BIG", Quantity: qty(1), UnitPrice: usd(100000)}},
        Total:      usd(100000), // => Above any tier's L2 threshold — triggers L3 (CFO) approval.
    }
}
 
// => InMemoryPORepo returns a ready-to-use fake repository pre-seeded with pos.
func NewInMemoryPORepo(pos ...PurchaseOrder) *InMemoryPORepository {
    repo := &InMemoryPORepository{store: make(map[string]PurchaseOrder)}
    for _, po := range pos {
        repo.store[po.ID] = po
    }
    return repo
}
 
// --- Test using the factories ---
 
func TestApprove_HighValuePO_RequiresCFO(t *testing.T) {
    // => Factory call communicates test intent in one line; no field setup noise.
    po := TestPOFactory{}.HighValuePO()
    policy := DefaultPolicyTable().Lookup("gold")
 
    // => Submit the pure function — no DB, no mocks, no network.
    submitted, err := Submit(po, policy)
    require.NoError(t, err)
 
    // => Assert that the high-value path requires L3 (CFO) approval.
    assert.Equal(t, ApprovalLevelL3, submitted.RequiredLevel)
}
 
func TestApprovalService_ApprovePO_SavesAndPublishes(t *testing.T) {
    // => Pre-seeded repo: no manual Save call needed in test setup.
    submitted := SubmittedPO{ID: "PO-001", Total: usd(1500), RequiredLevel: ApprovalLevelL1}
    repo := NewInMemoryPORepo()
 
    bus := NewEventBus()
    published := []DomainEvent{}
    // => Spy: capture published events for assertion without a mock framework.
    bus.Subscribe("po.approved", EventHandlerFunc(func(_ context.Context, e DomainEvent) error {
        published = append(published, e)
        return nil
    }))
 
    svc := NewApprovalService(repo, bus)
    err := svc.Approve(context.Background(), submitted.ID, "MGR-001")
    require.NoError(t, err)
 
    // => Assert state and behaviour in the same test without separate mock expectations.
    assert.Len(t, published, 1)
}