Echo Chamber #

// Echo chamber signs in traces:

// Healthy traces:
// Client → A (50ms) → B (20ms) → [done]

// Traces showing echo chambers:
// Client → A → B → A → B → A → B → ... (hundreds of spans from two services)

// In Jaeger/Zipkin: look for traces with:
// - Very high span counts from the same two services
// - Very deep nesting (endless nested calls)
// - Durations far exceeding expectations

// With OpenTelemetry, traces show this visually

package main

import (
	"bufio"
	"fmt"
	"os"
	"regexp"
)

// detectEchoChamberFromLogs detects echo chamber patterns from logs.
// If two services call each other more than N times within a time window,
// an echo chamber is likely happening.
func detectEchoChamberFromLogs(logFile string, windowSeconds int) {
	callPairs := map[string]int{}
	timeWindows := map[string]string{}

	logLine := regexp.MustCompile(`(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}) (\w+) -> (\w+) req=(\w+)`)

	f, err := os.Open(logFile)
	if err != nil {
		return
	}
	defer f.Close()

	scanner := bufio.NewScanner(f)
	for scanner.Scan() {
		line := scanner.Text()
		// Parse the log: timestamp, caller_service, callee_service, request_id
		match := logLine.FindStringSubmatch(line)
		if match == nil {
			continue
		}

		timestampStr, caller, callee := match[1], match[2], match[3]

		pair := caller + " ↔ " + callee
		reversePair := callee + " ↔ " + caller

		// Check whether the reverse pair exists within the time window
		if _, ok := timeWindows[reversePair]; ok {
			// If the pair calls each other within the window
			callPairs[pair]++
		}

		timeWindows[pair] = timestampStr
	}

	// Report suspicious pairs
	for pair, count := range callPairs {
		if count > 10 { // threshold
			fmt.Printf("⚠️  Possible echo chamber: %s (%d mutual calls)\n", pair, count)
		}
	}
}

package main

import (
	"github.com/prometheus/client_golang/prometheus"
)

// Prometheus metrics for detecting echo chambers

// Track every outbound request with source and destination
var outboundRequests = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "service_outbound_requests_total",
		Help: "Total outbound requests to other services",
	},
	[]string{"caller_service", "callee_service", "endpoint"},
)

// If services A and B call each other at high frequencies:
// Prometheus query:
// rate(service_outbound_requests_total{caller_service="service-a",callee_service="service-b"}[1m])
// AND
// rate(service_outbound_requests_total{caller_service="service-b",callee_service="service-a"}[1m])
// Both high simultaneously → an echo chamber signal

// Alert rule:
// alert: PossibleEchoChamber
// expr: |
//   (
//     rate(service_outbound_requests_total{caller_service="service-a", callee_service="service-b"}[2m])
//     > 10
//   ) and (
//     rate(service_outbound_requests_total{caller_service="service-b", callee_service="service-a"}[2m])
//     > 10
//   )
// for: 1m
// annotations:
//   summary: "Possible echo chamber between service-a and service-b"

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"net/http"
	"strconv"
	"time"
)

const maxRequestDepth = 5 // maximum 5 levels of request chaining

type ctxDepthKey struct{}

// checkRequestDepth checks and enforces request depth limits.
// Depth is propagated via the X-Request-Depth header.
func checkRequestDepth(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Get the depth from the header (sent by the caller)
		depth, err := strconv.Atoi(r.Header.Get("X-Request-Depth"))
		if err != nil {
			depth = 0
		}

		// Store the depth in the request context
		r = r.WithContext(context.WithValue(r.Context(), ctxDepthKey{}, depth))

		// Reject if already too deep
		if depth >= maxRequestDepth {
			w.Header().Set("Content-Type", "application/json")
			w.WriteHeader(508) // HTTP 508 Loop Detected
			fmt.Fprintf(w,
				`{"error":"Request depth limit exceeded","depth":%d,"max_depth":%d,"message":"Possible circular dependency detected"}`,
				depth, maxRequestDepth)
			return
		}

		next.ServeHTTP(w, r)
	})
}

// callService is a wrapper for HTTP calls to other services that increments depth.
// Use this for all inter-service calls, not httpx/requests directly.
func callService(r *http.Request, url string) (map[string]interface{}, error) {
	depth := r.Context().Value(ctxDepthKey{}).(int)
	requestID := r.Header.Get("X-Request-ID")
	if requestID == "" {
		requestID = newRequestID()
	}

	client := &http.Client{Timeout: 10 * time.Second}

	req, err := http.NewRequest(http.MethodGet, url, nil)
	if err != nil {
		return nil, err
	}
	req.Header.Set("X-Request-Depth", strconv.Itoa(depth+1)) // increment the depth
	req.Header.Set("X-Request-ID", requestID)                // propagate the ID for tracing
	req.Header.Set("X-Caller-Service", "service-a")          // identify the caller

	resp, err := client.Do(req)
	if err != nil {
		return nil, err
	}
	defer resp.Body.Close()

	var result map[string]interface{}
	if err := json.NewDecoder(resp.Body).Decode(&result); err != nil {
		return nil, err
	}
	return result, nil
}

// Usage: register the middleware on the router:
// mux.Handle("/orders", checkRequestDepth(http.HandlerFunc(createOrder)))

// createOrder calls other services with automatic depth tracking
func createOrder(w http.ResponseWriter, r *http.Request) {
	var orderData map[string]interface{}
	json.NewDecoder(r.Body).Decode(&orderData)

	// Call other services with automatic depth tracking
	user, err := callService(r, fmt.Sprintf("http://user-service/users/%v", orderData["user_id"]))
	if err != nil {
		http.Error(w, err.Error(), http.StatusBadGateway)
		return
	}

	json.NewEncoder(w).Encode(createOrderRecord(orderData, user))
}

package main

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"fmt"
	"sort"
	"time"

	"github.com/redis/go-redis/v9"
)

var redisClient = redis.NewClient(&redis.Options{
	Addr: "redis:6379",
})

// isEventProcessed checks whether an event with this ID has already been processed.
// Uses Redis to store processed event IDs.
func isEventProcessed(ctx context.Context, eventID string, ttlSeconds int) bool {
	key := fmt.Sprintf("processed_event:%s", eventID)

	// SET NX: only set if it doesn't exist (atomic).
	// Returns true if successfully set (the event has never been processed).
	wasSet, err := redisClient.SetNX(ctx, key, "1", time.Duration(ttlSeconds)*time.Second).Result()
	if err != nil {
		return false
	}
	return !wasSet // true = already processed (no need to process again)
}

// processEventIdempotent processes events with idempotency checks.
// If the event was already processed, skip without errors.
func processEventIdempotent(ctx context.Context, event map[string]interface{}) map[string]interface{} {
	eventID, _ := event["event_id"].(string)
	if eventID == "" {
		// Generate an event ID from the content if missing
		pairs := make([]string, 0, len(event))
		for k, v := range event {
			pairs = append(pairs, fmt.Sprintf("%v:%v", k, v))
		}
		sort.Strings(pairs)

		h := sha256.New()
		fmt.Fprintf(h, "%v", pairs)
		eventID = hex.EncodeToString(h.Sum(nil))[:16]
	}

	if isEventProcessed(ctx, eventID, 3600) {
		logger.Info("Event already processed, skipping",
			"event_id", eventID, "event_type", event["type"])
		return map[string]interface{}{"status": "already_processed", "event_id": eventID}
	}

	// Process the event
	result := handleEvent(event)
	logger.Info("Event processed successfully",
		"event_id", eventID, "event_type", event["type"])
	return result
}

// For webhooks: always use the provider's webhook ID
func paymentWebhook(w http.ResponseWriter, r *http.Request) {
	webhookID := r.Header.Get("X-Webhook-ID")
	if webhookID == "" {
		webhookID = r.FormValue("id")
	}

	if webhookID == "" {
		http.Error(w, `{"error": "Missing webhook ID"}`, http.StatusBadRequest)
		return
	}

	// Idempotency check
	if isEventProcessed(r.Context(), fmt.Sprintf("webhook:%s", webhookID), 3600) {
		w.WriteHeader(http.StatusOK)
		fmt.Fprint(w, `{"status": "already_processed"}`)
		return
	}

	// Process the webhook
	processPaymentEvent(r.Body)
	w.WriteHeader(http.StatusOK)
	fmt.Fprint(w, `{"status": "processed"}`)
}

package main

import (
	"fmt"
	"time"
)

// An event envelope storing the chain of causality
type EventEnvelope struct {
	EventID       string
	EventType     string
	Payload       map[string]interface{}
	CorrelationID string // the same for all related events
	CausationID   string // the ID of the event that caused this event
	CauseChain    []string
	SourceService string
	Timestamp     time.Time
}

const maxCauseChainDepth = 10

// NewEventEnvelope creates a new event envelope with fresh IDs.
func NewEventEnvelope(eventType string, payload map[string]interface{}, source string) *EventEnvelope {
	return &EventEnvelope{
		EventID:       newEventID(),
		EventType:     eventType,
		Payload:       payload,
		CorrelationID: newCorrelationID(),
		CauseChain:    []string{},
		SourceService: source,
		Timestamp:     time.Now(),
	}
}

// CreateChildEvent creates a new event resulting from this event.
func (e *EventEnvelope) CreateChildEvent(eventType string, payload map[string]interface{}, source string) (*EventEnvelope, error) {
	if len(e.CauseChain) >= maxCauseChainDepth {
		return nil, fmt.Errorf(
			"cause chain too deep (%d). possible echo chamber. chain: %v",
			len(e.CauseChain), e.CauseChain)
	}

	return &EventEnvelope{
		EventType:     eventType,
		Payload:       payload,
		CorrelationID: e.CorrelationID,                 // the same for all related events
		CausationID:   e.EventID,                       // the ID of the causing event
		CauseChain:    append(e.CauseChain, e.EventID), // add to the chain
		SourceService: source,
		Timestamp:     time.Now(),
	}, nil
}

// EchoChamberDetected signals a cause chain that is too deep.
type EchoChamberDetected struct{ msg string }

func (e *EchoChamberDetected) Error() string { return e.msg }

// Echo-chamber-aware consumers
func consumeUserUpdatedEvent(envelope *EventEnvelope) {
	userID := envelope.Payload["user_id"]

	// Check whether this event is already in a chain too deep
	if containsEventType(envelope.CauseChain, "user.profile.sync") {
		logger.Warning("Skipping user profile sync — already in cause chain",
			"cause_chain", envelope.CauseChain)
		return
	}

	// Update the local profile
	updateLocalProfile(userID, envelope.Payload)

	// If other events need triggering, create them as child events
	child, err := envelope.CreateChildEvent(
		"user.profile.sync",
		map[string]interface{}{"user_id": userID, "synced_at": time.Now().Unix()},
		"profile-service",
	)
	if err != nil {
		logger.Error("Echo chamber prevented: %v", err)
		// Stop the chain — don't publish events that would create loops
		return
	}
	publishEvent(child)
}

package main

// Refactor example: from circular dependencies to event-driven

// BEFORE (circular):
// Order Service
func createOrderCircular(orderData map[string]interface{}) {
	// Calls the User Service for validation
	user := userService.GetUser(orderData["user_id"]) // → User Service
	// The User Service also calls the Order Service to check order limits
	// → CIRCULAR!
	_ = user
}

// User Service
func getUserCircular(userID string) {
	user := db.GetUser(userID)
	// Check the order limit
	orders := orderService.GetOrders(userID) // → Order Service → CIRCULAR!
	user["can_order"] = len(orders) < user["order_limit"]
}

// AFTER (event-driven, no circularity):
// Order Service — only knows about orders
func createOrder(orderData map[string]interface{}) {
	// Doesn't call the User Service!
	// Order limit info already exists in the payload (sent by clients/API gateways)
	if orderData["user_order_count"].(int) >= orderData["user_order_limit"].(int) {
		panic(OrderLimitExceeded{})
	}

	order := newOrder(orderData)
	db.Save(order)

	// Publish events — no need to know who consumes them
	publishEvent("order.created", map[string]interface{}{"order_id": order.ID, "user_id": order.UserID})
}

// User Service — only knows about users
func getUser(userID string) {
	// Doesn't call the Order Service!
	// Only returns user data existing in its own domain
	_ = db.GetUser(userID)
}

// API Gateways or BFFs — the aggregators
func checkout(request map[string]interface{}) {
	user := userService.GetUser(request["user_id"])
	orderCount := orderService.CountOrders(request["user_id"])

	// Combine data at the upper layer, not inside services
	orderService.CreateOrder(map[string]interface{}{
		"user_order_count": orderCount,
		"user_order_limit": user["order_limit"],
	})
}

package main

import (
	"fmt"
	"time"
)

// A minimal circuit breaker for inter-service calls
type CircuitBreaker struct {
	Name             string
	FailureThreshold int
	RecoveryTimeout  time.Duration
}

// CircuitOpenError is returned when the circuit is open.
type CircuitOpenError struct{ service string }

func (e *CircuitOpenError) Error() string {
	return fmt.Sprintf("circuit open for %s", e.service)
}

// Call runs fn, counting failures against the threshold.
func (cb *CircuitBreaker) Call(fn func() (interface{}, error)) (interface{}, error) {
	if cb.isOpen() {
		return nil, &CircuitOpenError{service: cb.Name}
	}
	result, err := fn()
	if err != nil {
		cb.recordFailure()
	}
	return result, err
}

// Circuit breakers on every inter-service call
var (
	userServiceBreaker  = &CircuitBreaker{Name: "user-service", FailureThreshold: 5, RecoveryTimeout: 30 * time.Second}
	orderServiceBreaker = &CircuitBreaker{Name: "order-service", FailureThreshold: 5, RecoveryTimeout: 30 * time.Second}
)

func getUserSafe(userID int) map[string]interface{} {
	result, err := userServiceBreaker.Call(func() (interface{}, error) {
		// pseudo HTTP call
		return httpGetJSON(fmt.Sprintf("http://user-service/users/%d", userID))
	})
	if err != nil {
		// The User Service is unavailable or in an echo chamber.
		// Return minimal data or raise a clear error.
		if _, ok := err.(*CircuitOpenError); ok {
			return map[string]interface{}{"id": userID, "status": "unavailable"}
		}
		return nil
	}
	return result.(map[string]interface{})
}

// With circuit breakers:
// If an echo chamber happens and calls start failing,
// circuits open after 5 failures
// → Requests fail fast instead of looping forever
// → Dependencies get time to recover

package main

import (
	"context"
	"net/http"
	"strconv"

	"github.com/prometheus/client_golang/prometheus"
)

// Custom metrics for early echo chamber detection

// Track request depth distributions
var requestDepthHistogram = prometheus.NewHistogram(prometheus.HistogramOpts{
	Name:    "http_request_depth",
	Help:    "Distribution of request depth (how deep in the call chain)",
	Buckets: []float64{0, 1, 2, 3, 5, 10},
})

// Alert if requests with high depths exist
var requestDepthExceeded = prometheus.NewCounter(prometheus.CounterOpts{
	Name: "request_depth_limit_exceeded_total",
	Help: "Number of requests rejected due to depth limits",
})

// Track mutual calls between services
var mutualCalls = prometheus.NewCounterVec(prometheus.CounterOpts{
	Name: "inter_service_mutual_calls_total",
	Help: "Number of times service A called B while processing a call from B",
}, []string{"service_a", "service_b"})

// trackRequestDepth records the depth and the caller of each request.
func trackRequestDepth(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		depth, _ := strconv.Atoi(r.Header.Get("X-Request-Depth"))
		requestDepthHistogram.Observe(float64(depth))

		// Store the depth in the request context
		r = r.WithContext(context.WithValue(r.Context(), "request_depth", depth))

		caller := r.Header.Get("X-Caller-Service")
		if caller != "" && caller != "service-a" {
			// Track that another service called us
			mutualCalls.WithLabelValues(caller, "service-a").Inc()
		}

		next.ServeHTTP(w, r)
	})
}

package main

// ✗ Anti-pattern 1: no request depth limits
// Services calling each other without bounds
// One request can produce thousands of calls before timeouts
// ✓ Solution: X-Request-Depth headers + maximum limits

// ✗ Anti-pattern 2: webhooks without idempotency checks
func paymentWebhook(w http.ResponseWriter, r *http.Request) {
	processPayment(r.Body) // can be processed repeatedly!
	w.Write([]byte(`{"ok": true}`))
}
// ✓ Solution: store processed webhook IDs in Redis

// ✗ Anti-pattern 3: event consumers publishing the same event types
func handleUserUpdated(event Event) {
	updateProfile(event.UserID)
	publish("user.updated", event) // will be consumed by itself again!
}
// ✓ Solution: event consumers don't publish the same events they consume
// or: check source services before publishing

// ✗ Anti-pattern 4: depending on each other without abstractions
// Service A directly imports Service B's clients
// Service B directly imports Service A's clients
// ✓ Solution: redesign with shared dependencies or event-driven approaches

// ✗ Anti-pattern 5: no timeouts on inter-service calls
func antiPattern5() {
	response := httpGet("http://service-b/data") // can hang forever
	// ✓ Solution: always set timeouts
	response = httpGetWithTimeout("http://service-b/data", 10*time.Second)
}