Echo Chamber #

// Tanda-tanda echo chamber di trace:

// Trace yang sehat:
// Client → A (50ms) → B (20ms) → [done]

// Trace yang menunjukkan echo chamber:
// Client → A → B → A → B → A → B → ... (ratusan span dari dua service)

// Di Jaeger/Zipkin: cari trace dengan:
// - Span count yang sangat tinggi dari dua service yang sama
// - Depth yang sangat dalam (nested calls yang tidak berakhir)
// - Duration yang jauh melebihi ekspektasi

// Dengan OpenTelemetry, trace menunjukkan ini secara visual

package main

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

// detectEchoChamberFromLogs mendeteksi pola echo chamber dari log.
// Jika dua service saling memanggil lebih dari N kali dalam window waktu,
// kemungkinan terjadi echo chamber.
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 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

		// Cek apakah pasangan yang berlawanan ada dalam window waktu
		if _, ok := timeWindows[reversePair]; ok {
			// Jika pasangan saling memanggil dalam window
			callPairs[pair]++
		}

		timeWindows[pair] = timestampStr
	}

	// Report pasangan yang mencurigakan
	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 untuk mendeteksi echo chamber

// Track setiap outbound request dengan sumber dan tujuan
var outboundRequests = prometheus.NewCounterVec(
	prometheus.CounterOpts{
		Name: "service_outbound_requests_total",
		Help: "Total outbound requests to other services",
	},
	[]string{"caller_service", "callee_service", "endpoint"},
)

// Jika service A dan B saling memanggil dengan frekuensi tinggi:
// Query Prometheus:
// rate(service_outbound_requests_total{caller_service="service-a",callee_service="service-b"}[1m])
// DAN
// rate(service_outbound_requests_total{caller_service="service-b",callee_service="service-a"}[1m])
// Keduanya tinggi secara bersamaan → sinyal echo chamber

// 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 // maksimum 5 level request chaining

type ctxDepthKey struct{}

// checkRequestDepth mengecek dan menegakkan request depth limit.
// Depth di-propagate via header X-Request-Depth.
func checkRequestDepth(next http.Handler) http.Handler {
	return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
		// Ambil depth dari header (dikirim oleh caller)
		depth, err := strconv.Atoi(r.Header.Get("X-Request-Depth"))
		if err != nil {
			depth = 0
		}

		// Simpan depth di request context
		r = r.WithContext(context.WithValue(r.Context(), ctxDepthKey{}, depth))

		// Tolak jika sudah terlalu dalam
		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 adalah wrapper untuk HTTP call ke service lain yang meng-increment depth.
// Gunakan ini di semua inter-service call, bukan httpx/requests langsung.
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 depth
	req.Header.Set("X-Request-ID", requestID)                // propagate ID untuk tracing
	req.Header.Set("X-Caller-Service", "service-a")          // identify 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
}

// Penggunaan: daftarkan middleware di router:
// mux.Handle("/orders", checkRequestDepth(http.HandlerFunc(createOrder)))

// createOrder memanggil service lain dengan automatic depth tracking
func createOrder(w http.ResponseWriter, r *http.Request) {
	var orderData map[string]interface{}
	json.NewDecoder(r.Body).Decode(&orderData)

	// Call ke service lain dengan 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 mengecek apakah event dengan ID ini sudah pernah diproses.
// Menggunakan Redis untuk menyimpan event ID yang sudah diproses.
func isEventProcessed(ctx context.Context, eventID string, ttlSeconds int) bool {
	key := fmt.Sprintf("processed_event:%s", eventID)

	// SET NX: set hanya jika belum ada (atomic).
	// Return true jika berhasil di-set (event belum pernah diproses).
	wasSet, err := redisClient.SetNX(ctx, key, "1", time.Duration(ttlSeconds)*time.Second).Result()
	if err != nil {
		return false
	}
	return !wasSet // true = sudah pernah diproses (tidak perlu proses lagi)
}

// processEventIdempotent memproses event dengan idempotency check.
// Jika event sudah pernah diproses, skip tanpa error.
func processEventIdempotent(ctx context.Context, event map[string]interface{}) map[string]interface{} {
	eventID, _ := event["event_id"].(string)
	if eventID == "" {
		// Generate event ID dari konten jika tidak ada
		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}
	}

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

// Untuk webhook: selalu gunakan webhook ID dari provider
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
	}

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

package main

import (
	"fmt"
	"time"
)

// Event envelope yang menyimpan chain of causality
type EventEnvelope struct {
	EventID       string
	EventType     string
	Payload       map[string]interface{}
	CorrelationID string // sama untuk semua event terkait
	CausationID   string // ID dari event yang menyebabkan event ini
	CauseChain    []string
	SourceService string
	Timestamp     time.Time
}

const maxCauseChainDepth = 10

// NewEventEnvelope membuat event envelope baru dengan ID baru.
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(),
	}
}

// Buat event baru yang merupakan akibat dari event ini.
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,                 // sama untuk semua event terkait
		CausationID:   e.EventID,                       // ID event penyebab
		CauseChain:    append(e.CauseChain, e.EventID), // tambah ke chain
		SourceService: source,
		Timestamp:     time.Now(),
	}, nil
}

// EchoChamberDetected menandakan cause chain yang terlalu dalam.
type EchoChamberDetected struct{ msg string }

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

// Consumer yang aware terhadap echo chamber
func consumeUserUpdatedEvent(envelope *EventEnvelope) {
	userID := envelope.Payload["user_id"]

	// Cek apakah event ini sudah dalam chain yang terlalu dalam
	if containsEventType(envelope.CauseChain, "user.profile.sync") {
		logger.Warning("Skipping user profile sync — already in cause chain",
			"cause_chain", envelope.CauseChain)
		return
	}

	// Update profil lokal
	updateLocalProfile(userID, envelope.Payload)

	// Kalau perlu trigger event lain, buat sebagai child event
	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 chain — jangan publish event yang akan membuat loop
		return
	}
	publishEvent(child)
}

package main

// Contoh refactor: dari circular dependency ke event-driven

// SEBELUM (circular):
// Order Service
func createOrderCircular(orderData map[string]interface{}) {
	// Panggil User Service untuk validasi
	user := userService.GetUser(orderData["user_id"]) // → User Service
	// User Service juga panggil Order Service untuk cek limit order
	// → CIRCULAR!
	_ = user
}

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

// SESUDAH (event-driven, tanpa circular):
// Order Service — hanya tahu tentang order
func createOrder(orderData map[string]interface{}) {
	// Tidak memanggil User Service!
	// Order limit info sudah ada di payload (dikirim oleh client/API gateway)
	if orderData["user_order_count"].(int) >= orderData["user_order_limit"].(int) {
		panic(OrderLimitExceeded{})
	}

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

	// Publish event — tidak perlu tahu siapa yang consume
	publishEvent("order.created", map[string]interface{}{"order_id": order.ID, "user_id": order.UserID})
}

// User Service — hanya tahu tentang user
func getUser(userID string) {
	// Tidak memanggil Order Service!
	// Hanya return data user yang ada di domain-nya sendiri
	_ = db.GetUser(userID)
}

// API Gateway atau BFF — yang mengagregasi
func checkout(request map[string]interface{}) {
	user := userService.GetUser(request["user_id"])
	orderCount := orderService.CountOrders(request["user_id"])

	// Combine data di layer atas, bukan di dalam service
	orderService.CreateOrder(map[string]interface{}{
		"user_order_count": orderCount,
		"user_order_limit": user["order_limit"],
	})
}

package main

import (
	"fmt"
	"time"
)

// Circuit breaker minimal untuk inter-service call
type CircuitBreaker struct {
	Name             string
	FailureThreshold int
	RecoveryTimeout  time.Duration
}

// CircuitOpenError dikembalikan ketika circuit terbuka.
type CircuitOpenError struct{ service string }

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

// Call menjalankan fn, menghitung kegagalan terhadap 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 breaker di setiap 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 {
		// User Service tidak tersedia atau sedang dalam echo chamber.
		// Kembalikan data minimal atau raise error yang jelas.
		if _, ok := err.(*CircuitOpenError); ok {
			return map[string]interface{}{"id": userID, "status": "unavailable"}
		}
		return nil
	}
	return result.(map[string]interface{})
}

// Dengan circuit breaker:
// Jika echo chamber terjadi dan call mulai gagal,
// circuit terbuka setelah 5 kegagalan
// → Request langsung gagal (fast fail) alih-alih loop terus
// → Dependency punya waktu untuk recover

package main

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

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

// Custom metrics untuk mendeteksi echo chamber lebih awal

// Track request depth distribution
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 jika ada request dengan depth tinggi
var requestDepthExceeded = prometheus.NewCounter(prometheus.CounterOpts{
	Name: "request_depth_limit_exceeded_total",
	Help: "Number of requests rejected due to depth limits",
})

// Track mutual calls antar service
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 mencatat depth dan caller dari setiap 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))

		// Simpan depth di 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 bahwa service lain memanggil kita
			mutualCalls.WithLabelValues(caller, "service-a").Inc()
		}

		next.ServeHTTP(w, r)
	})
}

package main

// ✗ Anti-pattern 1: Tidak ada request depth limit
// Service yang saling memanggil tanpa batas
// Satu request bisa menghasilkan ribuan call sebelum timeout
// ✓ Solusi: X-Request-Depth header + batas maksimum

// ✗ Anti-pattern 2: Webhook tanpa idempotency check
func paymentWebhook(w http.ResponseWriter, r *http.Request) {
	processPayment(r.Body) // bisa diproses berkali-kali!
	w.Write([]byte(`{"ok": true}`))
}
// ✓ Solusi: simpan webhook ID yang sudah diproses di Redis

// ✗ Anti-pattern 3: Event consumer yang publish event dengan type yang sama
func handleUserUpdated(event Event) {
	updateProfile(event.UserID)
	publish("user.updated", event) // akan di-consume oleh dirinya sendiri lagi!
}
// ✓ Solusi: event consumer tidak publish event yang sama dengan yang di-consume
// atau: cek source service sebelum publish

// ✗ Anti-pattern 4: Saling bergantung tanpa abstraksi
// Service A import langsung client Service B
// Service B import langsung client Service A
// ✓ Solusi: redesain dengan shared dependency atau event-driven

// ✗ Anti-pattern 5: Tidak ada timeout pada inter-service call
func antiPattern5() {
	response := httpGet("http://service-b/data") // bisa hang selamanya
	// ✓ Solusi: selalu set timeout
	response = httpGetWithTimeout("http://service-b/data", 10*time.Second)
}