Intermediate

package main

import (
    "encoding/json"
    "fmt"
    "sync"
    "time"
)

// PubSubBroker: central coordinator holding topic -> subscriber mappings
type PubSubBroker struct {
    subscribers map[string][]func(map[string]interface{})
    mu          sync.Mutex
}

func NewPubSubBroker() *PubSubBroker {
    return &PubSubBroker{
        subscribers: make(map[string][]func(map[string]interface{})),
        // => subscribers is {} initially
    }
}

func (b *PubSubBroker) Subscribe(topic string, handler func(map[string]interface{})) {
    // Acquire lock before mutating shared subscriber map
    // => prevents race conditions when multiple services subscribe concurrently
    b.mu.Lock()
    defer b.mu.Unlock()
    b.subscribers[topic] = append(b.subscribers[topic], handler)
    // => handler is appended to topic's subscriber list
    fmt.Printf("[BROKER] Subscribed handler to topic '%s'\n", topic)
}

func (b *PubSubBroker) Publish(topic string, message map[string]interface{}) {
    // Snapshot subscriber list under lock to avoid holding lock during handler execution
    // => handlers can themselves call Subscribe without deadlocking
    b.mu.Lock()
    handlers := make([]func(map[string]interface{}), len(b.subscribers[topic]))
    copy(handlers, b.subscribers[topic])
    // => handlers is a shallow copy of the list at publish time
    b.mu.Unlock()

    jsonBytes, _ := json.Marshal(message)
    fmt.Printf("[BROKER] Publishing to '%s': %s\n", topic, string(jsonBytes))

    // Deliver to each subscriber independently — one slow handler doesn't block others
    var wg sync.WaitGroup
    for _, handler := range handlers {
        wg.Add(1)
        // Each handler runs in a separate goroutine simulating async delivery
        // => production systems use worker pools or async event loops instead
        go func(h func(map[string]interface{})) {
            defer wg.Done()
            h(message)
            // => message delivered concurrently to each subscriber
        }(handler)
    }
    wg.Wait()
}

// Subscriber services — each receives its own copy of the event
func emailService(message map[string]interface{}) {
    // Receives order.created event and sends confirmation email
    // => processes independently; failure here doesn't affect inventory service
    orderID := message["order_id"].(string)
    email := message["customer_email"].(string)
    fmt.Printf("[EMAIL]     Sending confirmation for order %s to %s\n", orderID, email)
}

func inventoryService(message map[string]interface{}) {
    // Decrements stock for each item in the order
    // => processes independently; can retry without affecting email service
    items := message["items"].([]interface{})
    for _, item := range items {
        m := item.(map[string]interface{})
        fmt.Printf("[INVENTORY] Reserving %.0f x %s\n", m["qty"].(float64), m["sku"].(string))
    }
}

func analyticsService(message map[string]interface{}) {
    // Records the order event for reporting dashboards
    // => decoupled from order placement; dashboard latency doesn't affect checkout
    fmt.Printf("[ANALYTICS] Recorded order %s worth $%.2f\n",
        message["order_id"].(string), message["total"].(float64))
}

func main() {
    // Wire up the broker
    broker := NewPubSubBroker()
    // => broker.subscribers is {}

    broker.Subscribe("order.created", emailService)
    // => broker.subscribers["order.created"] = [emailService]

    broker.Subscribe("order.created", inventoryService)
    // => broker.subscribers["order.created"] = [emailService, inventoryService]

    broker.Subscribe("order.created", analyticsService)
    // => broker.subscribers["order.created"] = [emailService, inventoryService, analyticsService]

    // Producer publishes a single event — all three subscribers receive a copy
    orderEvent := map[string]interface{}{
        "order_id":       "ORD-7821",
        "customer_email": "alice@example.com",
        "items": []interface{}{
            map[string]interface{}{"sku": "WIDGET-A", "qty": float64(2)},
            map[string]interface{}{"sku": "GADGET-B", "qty": float64(1)},
        },
        "total": 149.99,
    }
    broker.Publish("order.created", orderEvent)
    // => Output (order may vary due to goroutines):
    // [BROKER] Publishing to 'order.created': {...}
    // [EMAIL]     Sending confirmation for order ORD-7821 to alice@example.com
    // [INVENTORY] Reserving 2 x WIDGET-A
    // [INVENTORY] Reserving 1 x GADGET-B
    // [ANALYTICS] Recorded order ORD-7821 worth $149.99

    time.Sleep(100 * time.Millisecond) // allow goroutines to finish
}

import json
import threading
from collections import defaultdict
from typing import Callable

# PubSub broker: central coordinator holding topic -> subscriber mappings
class PubSubBroker:
    def __init__(self):
        # defaultdict(list) auto-creates empty list for unknown topics
        # => self._subscribers is {} initially
        self._subscribers: dict[str, list[Callable]] = defaultdict(list)
        # Lock ensures thread-safe subscribe/publish across goroutines
        # => self._lock is a reentrant threading lock
        self._lock = threading.Lock()

    def subscribe(self, topic: str, handler: Callable) -> None:
        # Acquire lock before mutating shared subscriber map
        # => prevents race conditions when multiple services subscribe concurrently
        with self._lock:
            self._subscribers[topic].append(handler)
            # => handler is appended to topic's subscriber list
            print(f"[BROKER] Subscribed handler to topic '{topic}'")

    def publish(self, topic: str, message: dict) -> None:
        # Snapshot subscriber list under lock to avoid holding lock during handler execution
        # => handlers can themselves call subscribe without deadlocking
        with self._lock:
            handlers = list(self._subscribers[topic])
            # => handlers is a shallow copy of the list at publish time

        print(f"[BROKER] Publishing to '{topic}': {message}")
        # Deliver to each subscriber independently — one slow handler doesn't block others
        for handler in handlers:
            # Each handler runs in a separate thread simulating async delivery
            # => production systems use worker pools or async event loops instead
            t = threading.Thread(target=handler, args=(message,), daemon=True)
            t.start()
            # => message delivered concurrently to each subscriber


# Subscriber services — each receives its own copy of the event
def email_service(message: dict) -> None:
    # Receives order.created event and sends confirmation email
    # => processes independently; failure here doesn't affect inventory service
    order_id = message["order_id"]
    email = message["customer_email"]
    print(f"[EMAIL]     Sending confirmation for order {order_id} to {email}")

def inventory_service(message: dict) -> None:
    # Decrements stock for each item in the order
    # => processes independently; can retry without affecting email service
    for item in message["items"]:
        print(f"[INVENTORY] Reserving {item['qty']} x {item['sku']}")

def analytics_service(message: dict) -> None:
    # Records the order event for reporting dashboards
    # => decoupled from order placement; dashboard latency doesn't affect checkout
    print(f"[ANALYTICS] Recorded order {message['order_id']} "
          f"worth ${message['total']:.2f}")


# Wire up the broker
broker = PubSubBroker()
# => broker._subscribers is {}

broker.subscribe("order.created", email_service)
# => broker._subscribers["order.created"] = [email_service]

broker.subscribe("order.created", inventory_service)
# => broker._subscribers["order.created"] = [email_service, inventory_service]

broker.subscribe("order.created", analytics_service)
# => broker._subscribers["order.created"] = [email_service, inventory_service, analytics_service]

# Producer publishes a single event — all three subscribers receive a copy
order_event = {
    "order_id": "ORD-7821",
    "customer_email": "alice@example.com",
    "items": [{"sku": "WIDGET-A", "qty": 2}, {"sku": "GADGET-B", "qty": 1}],
    "total": 149.99,
}
broker.publish("order.created", order_event)
# => Output (order may vary due to threading):
# [BROKER] Publishing to 'order.created': {...}
# [EMAIL]     Sending confirmation for order ORD-7821 to alice@example.com
# [INVENTORY] Reserving 2 x WIDGET-A
# [INVENTORY] Reserving 1 x GADGET-B
# [ANALYTICS] Recorded order ORD-7821 worth $149.99

import time; time.sleep(0.1)  # allow daemon threads to finish before exit

package main

import (
    "fmt"
    "math/rand"
    "sync"
    "time"
)

// WorkQueue wraps Go's buffered channel with producer/consumer helpers
type WorkQueue struct {
    ch        chan map[string]string
    processed int
    mu        sync.Mutex
}

func NewWorkQueue(maxSize int) *WorkQueue {
    return &WorkQueue{
        // Buffered channel provides natural backpressure
        // => maxSize=20 means send blocks when 20 items are queued
        ch: make(chan map[string]string, maxSize),
    }
}

func (wq *WorkQueue) Enqueue(task map[string]string) {
    // Send blocks if channel is full — provides natural backpressure to producers
    // => producer slows down automatically when workers can't keep up
    wq.ch <- task
    fmt.Printf("[QUEUE] Enqueued task %s (qsize~=%d)\n", task["id"], len(wq.ch))
}

func (wq *WorkQueue) StartWorker(workerID int, wg *sync.WaitGroup) {
    // Each worker runs range on channel — exactly one worker receives each item
    // => competitive consumption: workers race to acquire each task
    go func() {
        defer wg.Done()
        for task := range wq.ch {
            // Simulate variable processing time (e.g., image resize, PDF render)
            duration := time.Duration(50+rand.Intn(100)) * time.Millisecond
            time.Sleep(duration)
            // => only one worker processes this task; others process different tasks

            wq.mu.Lock()
            wq.processed++
            count := wq.processed
            wq.mu.Unlock()
            fmt.Printf("[WORKER-%d] Processed task %s in %dms (total=%d)\n",
                workerID, task["id"], duration.Milliseconds(), count)
        }
    }()
}

func (wq *WorkQueue) Shutdown() {
    // Closing the channel signals all workers to exit when drained
    // => range loop exits when channel is closed and empty
    close(wq.ch)
}

func main() {
    numWorkers := 3
    wq := NewWorkQueue(20)
    // => wq.ch is empty, bounded at 20 items

    // Start three competing workers — they share the same channel
    var wg sync.WaitGroup
    for i := 0; i < numWorkers; i++ {
        wg.Add(1)
        wq.StartWorker(i, &wg)
    }
    // => workers[0], workers[1], workers[2] all blocked on channel receive

    // Enqueue 9 tasks — distributed across 3 workers, ~3 tasks each
    for i := 0; i < 9; i++ {
        wq.Enqueue(map[string]string{
            "id":      fmt.Sprintf("TASK-%03d", i),
            "payload": fmt.Sprintf("data_%d", i),
        })
    }
    // => Output example (worker assignment varies):
    // [QUEUE] Enqueued task TASK-000 (qsize=1)
    // [WORKER-0] Processed task TASK-000 in 80ms (total=1)
    // [WORKER-1] Processed task TASK-001 in 120ms (total=2)
    // ... (each task processed by exactly one worker)

    wq.Shutdown()
    wg.Wait()
    // => all workers exit via channel close, Wait returns when all done
    fmt.Printf("[DONE] All tasks processed. Total: %d\n", wq.processed)
    // => Output: [DONE] All tasks processed. Total: 9
}

import queue
import threading
import time
import random

# WorkQueue wraps Python's thread-safe Queue with producer/consumer helpers
class WorkQueue:
    def __init__(self, max_size: int = 100):
        # queue.Queue is thread-safe FIFO with optional bounded size
        # => max_size=100 means put() blocks when 100 items are queued (backpressure)
        self._q: queue.Queue = queue.Queue(maxsize=max_size)
        self._processed = 0
        self._lock = threading.Lock()

    def enqueue(self, task: dict) -> None:
        # put() blocks if queue is full — provides natural backpressure to producers
        # => producer slows down automatically when workers can't keep up
        self._q.put(task)
        print(f"[QUEUE] Enqueued task {task['id']} (qsize={self._q.qsize()})")

    def start_worker(self, worker_id: int) -> threading.Thread:
        # Each worker runs get() in a loop — exactly one worker unblocks per item
        # => competitive consumption: workers race to acquire each task
        def work_loop():
            while True:
                # get() blocks until an item is available — no busy-waiting
                # => CPU is released to OS when queue is empty
                task = self._q.get()
                if task is None:
                    # Poison pill: None signals worker to exit gracefully
                    # => clean shutdown without killing threads forcefully
                    self._q.task_done()
                    break

                # Simulate variable processing time (e.g., image resize, PDF render)
                duration = random.uniform(0.05, 0.15)
                time.sleep(duration)
                # => only one worker processes this task; others process different tasks

                with self._lock:
                    self._processed += 1
                print(f"[WORKER-{worker_id}] Processed task {task['id']} "
                      f"in {duration:.2f}s (total={self._processed})")
                # task_done() decrements internal counter used by join()
                self._q.task_done()

        t = threading.Thread(target=work_loop, daemon=True)
        t.start()
        return t

    def shutdown(self, num_workers: int) -> None:
        # Send one poison pill per worker so each exits exactly once
        # => avoids sending too few (workers hang) or too many (spurious exits)
        for _ in range(num_workers):
            self._q.put(None)
        self._q.join()
        # => join() blocks until all task_done() calls match all get() calls


NUM_WORKERS = 3
wq = WorkQueue(max_size=20)
# => wq._q is empty, bounded at 20 items

# Start three competing workers — they share the same queue
workers = [wq.start_worker(i) for i in range(NUM_WORKERS)]
# => workers[0], workers[1], workers[2] all blocked on get()

# Enqueue 9 tasks — distributed across 3 workers, ~3 tasks each
for i in range(9):
    wq.enqueue({"id": f"TASK-{i:03d}", "payload": f"data_{i}"})
# => Output example (worker assignment varies):
# [QUEUE] Enqueued task TASK-000 (qsize=1)
# [WORKER-0] Processed task TASK-000 in 0.08s (total=1)
# [WORKER-1] Processed task TASK-001 in 0.12s (total=2)
# [WORKER-2] Processed task TASK-002 in 0.06s (total=3)
# ... (each task processed by exactly one worker)

wq.shutdown(NUM_WORKERS)
# => all workers exit via poison pill, join() returns when queue drained
print(f"[DONE] All tasks processed. Total: {wq._processed}")
# => Output: [DONE] All tasks processed. Total: 9

package main

import (
    "fmt"
    "sync"
    "time"
)

type Record struct {
    ID      int
    Value   string
    Version int // monotonically increasing — detects replication order issues
}

// PrimaryNode: accepts writes and broadcasts to registered replicas
type PrimaryNode struct {
    nodeID   string
    store    map[int]*Record
    replicas []*ReplicaNode
    version  int
    mu       sync.Mutex
}

func NewPrimaryNode(nodeID string) *PrimaryNode {
    return &PrimaryNode{
        nodeID: nodeID,
        store:  make(map[int]*Record),
    }
}

func (p *PrimaryNode) RegisterReplica(r *ReplicaNode) {
    // Replicas register at startup; primary pushes changes to all of them
    p.replicas = append(p.replicas, r)
    fmt.Printf("[%s] Replica '%s' registered\n", p.nodeID, r.nodeID)
}

func (p *PrimaryNode) Write(recordID int, value string) *Record {
    p.mu.Lock()
    p.version++
    record := &Record{ID: recordID, Value: value, Version: p.version}
    p.store[recordID] = record
    // => write committed to primary synchronously
    p.mu.Unlock()

    fmt.Printf("[%s] WRITE id=%d value='%s' version=%d\n",
        p.nodeID, recordID, value, record.Version)

    // Replicate asynchronously — primary doesn't wait for replicas to confirm
    // => this is the source of replica lag; primary returns immediately
    for _, replica := range p.replicas {
        go replica.applyReplication(record)
    }

    return record
}

func (p *PrimaryNode) Read(recordID int) *Record {
    // Primary reads are always consistent — no lag possible
    p.mu.Lock()
    defer p.mu.Unlock()
    return p.store[recordID]
}

// ReplicaNode: read-only; applies replication events from primary
type ReplicaNode struct {
    nodeID string
    store  map[int]*Record
    lagMs  float64 // Simulated replication lag in milliseconds
    mu     sync.Mutex
}

func NewReplicaNode(nodeID string, lagMs float64) *ReplicaNode {
    return &ReplicaNode{
        nodeID: nodeID,
        store:  make(map[int]*Record),
        // real MySQL/PostgreSQL lag is 1-500ms typically
        // => high-traffic primaries can push lag to seconds under write storms
        lagMs: lagMs,
    }
}

func (r *ReplicaNode) applyReplication(record *Record) {
    // Simulate network + disk write delay on replica
    time.Sleep(time.Duration(r.lagMs) * time.Millisecond)
    r.mu.Lock()
    r.store[record.ID] = record
    r.mu.Unlock()
    fmt.Printf("[%s] REPLICATED id=%d version=%d (lag=%.0fms)\n",
        r.nodeID, record.ID, record.Version, r.lagMs)
}

func (r *ReplicaNode) Read(recordID int) *Record {
    // Reads from replica — may return stale data if replication hasn't caught up
    r.mu.Lock()
    record := r.store[recordID]
    r.mu.Unlock()
    staleness := "FRESH"
    if record == nil {
        staleness = "STALE/MISSING"
    }
    fmt.Printf("[%s] READ id=%d -> %s\n", r.nodeID, recordID, staleness)
    return record
}

func main() {
    // Setup: one primary, two replicas with different lag characteristics
    primary := NewPrimaryNode("primary")
    replica1 := NewReplicaNode("replica-1", 30)
    replica2 := NewReplicaNode("replica-2", 80)

    primary.RegisterReplica(replica1)
    primary.RegisterReplica(replica2)
    // => [primary] Replica 'replica-1' registered
    // => [primary] Replica 'replica-2' registered

    // Write to primary
    primary.Write(1, "hello")
    // => [primary] WRITE id=1 value='hello' version=1

    // Immediately read from replica — may be stale!
    time.Sleep(10 * time.Millisecond) // 10ms — replica-1 lag is 30ms, so both are still behind
    replica1.Read(1)
    // => [replica-1] READ id=1 -> STALE/MISSING (replication hasn't arrived yet)

    time.Sleep(100 * time.Millisecond) // wait 100ms — both replicas have now caught up
    replica1.Read(1)
    // => [replica-1] READ id=1 -> FRESH
    // => replica-1 returns Record{ID:1, Value:"hello", Version:1}
}

import time
import threading
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class Record:
    id: int
    value: str
    version: int  # monotonically increasing — detects replication order issues

# PrimaryNode: accepts writes and broadcasts to registered replicas
class PrimaryNode:
    def __init__(self, node_id: str):
        self.node_id = node_id
        # In-memory store simulating the primary's data store
        self._store: dict[int, Record] = {}
        self._replicas: list["ReplicaNode"] = []
        self._version = 0
        self._lock = threading.Lock()

    def register_replica(self, replica: "ReplicaNode") -> None:
        # Replicas register at startup; primary pushes changes to all of them
        self._replicas.append(replica)
        print(f"[{self.node_id}] Replica '{replica.node_id}' registered")

    def write(self, record_id: int, value: str) -> Record:
        with self._lock:
            self._version += 1
            record = Record(id=record_id, value=value, version=self._version)
            self._store[record_id] = record
            # => write committed to primary synchronously

        print(f"[{self.node_id}] WRITE id={record_id} value='{value}' "
              f"version={record.version}")

        # Replicate asynchronously — primary doesn't wait for replicas to confirm
        # => this is the source of replica lag; primary returns immediately
        for replica in self._replicas:
            threading.Thread(
                target=replica._apply_replication,
                args=(record,),
                daemon=True
            ).start()

        return record

    def read(self, record_id: int) -> Optional[Record]:
        # Primary reads are always consistent — no lag possible
        with self._lock:
            return self._store.get(record_id)


# ReplicaNode: read-only; applies replication events from primary
class ReplicaNode:
    def __init__(self, node_id: str, replication_lag_ms: float = 50.0):
        self.node_id = node_id
        self._store: dict[int, Record] = {}
        # Simulated replication lag: real MySQL/PostgreSQL lag is 1-500ms typically
        # => high-traffic primaries can push lag to seconds under write storms
        self._lag_ms = replication_lag_ms
        self._lock = threading.Lock()

    def _apply_replication(self, record: Record) -> None:
        # Simulate network + disk write delay on replica
        time.sleep(self._lag_ms / 1000.0)
        with self._lock:
            self._store[record.id] = record
        print(f"[{self.node_id}] REPLICATED id={record.id} "
              f"version={record.version} (lag={self._lag_ms}ms)")

    def read(self, record_id: int) -> Optional[Record]:
        # Reads from replica — may return stale data if replication hasn't caught up
        with self._lock:
            record = self._store.get(record_id)
        staleness = "FRESH" if record else "STALE/MISSING"
        print(f"[{self.node_id}] READ id={record_id} -> {staleness}")
        return record


# Setup: one primary, two replicas with different lag characteristics
primary = PrimaryNode("primary")
replica1 = ReplicaNode("replica-1", replication_lag_ms=30)
replica2 = ReplicaNode("replica-2", replication_lag_ms=80)

primary.register_replica(replica1)
primary.register_replica(replica2)
# => [primary] Replica 'replica-1' registered
# => [primary] Replica 'replica-2' registered

# Write to primary
primary.write(1, "hello")
# => [primary] WRITE id=1 value='hello' version=1

# Immediately read from replica — may be stale!
time.sleep(0.01)  # 10ms — replica-1 lag is 30ms, so both replicas are still behind
r = replica1.read(1)
# => [replica-1] READ id=1 -> STALE/MISSING (replication hasn't arrived yet)

time.sleep(0.1)  # wait 100ms — both replicas have now caught up
r = replica1.read(1)
# => [replica-1] READ id=1 -> FRESH
# => replica-1 returns Record(id=1, value='hello', version=1)

package main

import (
    "fmt"
    "sync"
    "time"
)

type ConflictStrategy int

const (
    LastWriteWins   ConflictStrategy = iota // highest timestamp wins — simple, lossy
    HighestVersion                          // highest version counter wins
)

type VersionedRecord struct {
    ID        int
    Value     string
    Timestamp float64 // wall clock at write time — used for LWW
    NodeID    string  // which master originated this write
    Version   int     // logical clock increment per write
}

type MasterNode struct {
    nodeID   string
    strategy ConflictStrategy
    store    map[int]*VersionedRecord
    peers    []*MasterNode
    mu       sync.Mutex
}

func NewMasterNode(nodeID string, strategy ConflictStrategy) *MasterNode {
    return &MasterNode{
        nodeID:   nodeID,
        strategy: strategy,
        store:    make(map[int]*VersionedRecord),
    }
}

func (m *MasterNode) AddPeer(peer *MasterNode) {
    // Bidirectional peering: each master knows all other masters
    m.peers = append(m.peers, peer)
}

func (m *MasterNode) Write(recordID int, value string) *VersionedRecord {
    m.mu.Lock()
    current := m.store[recordID]
    newVersion := 1
    if current != nil {
        newVersion = current.Version + 1
    }
    record := &VersionedRecord{
        ID:        recordID,
        Value:     value,
        Timestamp: float64(time.Now().UnixNano()) / 1e9,
        NodeID:    m.nodeID,
        Version:   newVersion,
    }
    m.store[recordID] = record
    m.mu.Unlock()

    fmt.Printf("[%s] LOCAL WRITE id=%d value='%s' v=%d ts=%.4f\n",
        m.nodeID, recordID, value, record.Version, record.Timestamp)

    // Propagate to peers asynchronously — conflict possible if both wrote concurrently
    for _, peer := range m.peers {
        go peer.receiveReplication(record)
    }
    return record
}

func (m *MasterNode) receiveReplication(incoming *VersionedRecord) {
    // Simulate network delay — this is when conflicts become visible
    time.Sleep(50 * time.Millisecond)
    m.mu.Lock()
    defer m.mu.Unlock()

    existing := m.store[incoming.ID]
    if existing == nil {
        // No local record — apply unconditionally
        m.store[incoming.ID] = incoming
        fmt.Printf("[%s] REPLICATED id=%d from %s (no conflict)\n",
            m.nodeID, incoming.ID, incoming.NodeID)
        return
    }

    // CONFLICT: both nodes have a version of this record
    // => apply conflict resolution strategy
    winner := m.resolveConflict(existing, incoming)
    m.store[incoming.ID] = winner
    loserNode := incoming.NodeID
    if winner == incoming {
        loserNode = existing.NodeID
    }
    fmt.Printf("[%s] CONFLICT id=%d: winner='%s' from %s, DISCARDED write from %s\n",
        m.nodeID, incoming.ID, winner.Value, winner.NodeID, loserNode)
}

func (m *MasterNode) resolveConflict(local, incoming *VersionedRecord) *VersionedRecord {
    if m.strategy == LastWriteWins {
        // LWW: higher timestamp wins — assumes clocks are synchronized (NTP)
        // => clock skew can cause older writes to win; data loss is possible
        if local.Timestamp >= incoming.Timestamp {
            return local
        }
        return incoming
    }
    // Version-based: higher logical version wins
    // => version monotonicity only holds if both nodes started from same base
    if local.Version >= incoming.Version {
        return local
    }
    return incoming
}

func (m *MasterNode) Read(recordID int) *VersionedRecord {
    m.mu.Lock()
    defer m.mu.Unlock()
    return m.store[recordID]
}

func main() {
    // Two geographically distributed masters (e.g., US-East and EU-West)
    masterUS := NewMasterNode("master-us", LastWriteWins)
    masterEU := NewMasterNode("master-eu", LastWriteWins)
    masterUS.AddPeer(masterEU)
    masterEU.AddPeer(masterUS)

    // Concurrent writes to same record from different nodes — guaranteed conflict
    masterUS.Write(42, "US value")
    masterEU.Write(42, "EU value")
    // => [master-us] LOCAL WRITE id=42 value='US value' v=1 ts=...
    // => [master-eu] LOCAL WRITE id=42 value='EU value' v=1 ts=...

    time.Sleep(200 * time.Millisecond) // wait for replication to propagate
    // => [master-us] CONFLICT id=42: winner='EU value' from master-eu (or vice versa)
    // => Both masters converge to same value — eventual consistency achieved

    usResult := masterUS.Read(42)
    euResult := masterEU.Read(42)
    fmt.Printf("US reads: '%s', EU reads: '%s'\n", usResult.Value, euResult.Value)
    // => Both nodes converge to the same value (LWW picks one winner)
    // => Output: US reads: 'EU value', EU reads: 'EU value'  (or both 'US value')
}

import time
import threading
from dataclasses import dataclass
from typing import Optional
from enum import Enum

class ConflictStrategy(Enum):
    LAST_WRITE_WINS = "lww"      # highest timestamp wins — simple, lossy
    HIGHEST_VERSION = "version"  # highest version counter wins

@dataclass
class VersionedRecord:
    id: int
    value: str
    timestamp: float    # wall clock at write time — used for LWW
    node_id: str        # which master originated this write
    version: int        # logical clock increment per write

class MasterNode:
    def __init__(self, node_id: str, strategy: ConflictStrategy):
        self.node_id = node_id
        self._strategy = strategy
        self._store: dict[int, VersionedRecord] = {}
        self._peers: list["MasterNode"] = []
        self._lock = threading.Lock()

    def add_peer(self, peer: "MasterNode") -> None:
        # Bidirectional peering: each master knows all other masters
        self._peers.append(peer)

    def write(self, record_id: int, value: str) -> VersionedRecord:
        with self._lock:
            current = self._store.get(record_id)
            # Increment version relative to existing record, or start at 1
            new_version = (current.version + 1) if current else 1
            record = VersionedRecord(
                id=record_id,
                value=value,
                timestamp=time.time(),
                node_id=self.node_id,
                version=new_version,
            )
            self._store[record_id] = record
        print(f"[{self.node_id}] LOCAL WRITE id={record_id} "
              f"value='{value}' v={record.version} ts={record.timestamp:.4f}")

        # Propagate to peers asynchronously — conflict possible if both wrote concurrently
        for peer in self._peers:
            threading.Thread(
                target=peer._receive_replication,
                args=(record,),
                daemon=True,
            ).start()
        return record

    def _receive_replication(self, incoming: VersionedRecord) -> None:
        # Simulate network delay — this is when conflicts become visible
        time.sleep(0.05)
        with self._lock:
            existing = self._store.get(incoming.id)
            if existing is None:
                # No local record — apply unconditionally
                self._store[incoming.id] = incoming
                print(f"[{self.node_id}] REPLICATED id={incoming.id} "
                      f"from {incoming.node_id} (no conflict)")
                return

            # CONFLICT: both nodes have a version of this record
            # => apply conflict resolution strategy
            winner = self._resolve_conflict(existing, incoming)
            self._store[incoming.id] = winner
            loser_node = incoming.node_id if winner == existing else existing.node_id
            print(f"[{self.node_id}] CONFLICT id={incoming.id}: "
                  f"winner='{winner.value}' from {winner.node_id}, "
                  f"DISCARDED write from {loser_node}")

    def _resolve_conflict(
        self, local: VersionedRecord, incoming: VersionedRecord
    ) -> VersionedRecord:
        if self._strategy == ConflictStrategy.LAST_WRITE_WINS:
            # LWW: higher timestamp wins — assumes clocks are synchronized (NTP)
            # => clock skew can cause older writes to win; data loss is possible
            return local if local.timestamp >= incoming.timestamp else incoming
        else:  # HIGHEST_VERSION
            # Version-based: higher logical version wins
            # => version monotonicity only holds if both nodes started from same base
            return local if local.version >= incoming.version else incoming

    def read(self, record_id: int) -> Optional[VersionedRecord]:
        with self._lock:
            return self._store.get(record_id)


# Two geographically distributed masters (e.g., US-East and EU-West)
master_us = MasterNode("master-us", ConflictStrategy.LAST_WRITE_WINS)
master_eu = MasterNode("master-eu", ConflictStrategy.LAST_WRITE_WINS)
master_us.add_peer(master_eu)
master_eu.add_peer(master_us)

# Concurrent writes to same record from different nodes — guaranteed conflict
master_us.write(42, "US value")
master_eu.write(42, "EU value")
# => [master-us] LOCAL WRITE id=42 value='US value' v=1 ts=...
# => [master-eu] LOCAL WRITE id=42 value='EU value' v=1 ts=...

time.sleep(0.2)  # wait for replication to propagate
# => [master-us] CONFLICT id=42: winner='EU value' from master-eu (or vice versa)
# => Both masters converge to same value — eventual consistency achieved

us_result = master_us.read(42)
eu_result = master_eu.read(42)
print(f"US reads: '{us_result.value}', EU reads: '{eu_result.value}'")
# => Both nodes converge to the same value (LWW picks one winner)
# => Output: US reads: 'EU value', EU reads: 'EU value'  (or both 'US value')

package main

import (
    "fmt"
    "sort"
)

type Shard struct {
    ShardID    int
    RangeStart int // Inclusive lower bound of key range this shard owns
    RangeEnd   int // Inclusive upper bound of key range this shard owns
    store      map[int]map[string]string
}

func NewShard(shardID, rangeStart, rangeEnd int) *Shard {
    return &Shard{
        ShardID:    shardID,
        RangeStart: rangeStart,
        RangeEnd:   rangeEnd,
        store:      make(map[int]map[string]string),
    }
}

func (s *Shard) Insert(key int, value map[string]string) {
    s.store[key] = value
    fmt.Printf("[Shard-%d] INSERT key=%d range=[%d,%d]\n",
        s.ShardID, key, s.RangeStart, s.RangeEnd)
}

func (s *Shard) Get(key int) (map[string]string, bool) {
    v, ok := s.store[key]
    return v, ok
}

func (s *Shard) RangeScan(start, end int) []map[string]string {
    // Range scan is efficient: only keys in this shard's range are relevant
    // => no cross-shard communication needed for queries within one shard's range
    var results []map[string]string
    for k, v := range s.store {
        if k >= start && k <= end {
            results = append(results, v)
        }
    }
    return results
}

// RangeShardRouter: maps any key to its owning shard
type RangeShardRouter struct {
    shards      []*Shard
    breakpoints []int
}

func NewRangeShardRouter(shards []*Shard) *RangeShardRouter {
    // Shards must be sorted by RangeStart for binary search to work correctly
    sort.Slice(shards, func(i, j int) bool {
        return shards[i].RangeStart < shards[j].RangeStart
    })
    // Extract RangeStart values as sorted breakpoints for binary search
    // => sort.SearchInts(breakpoints, key) gives index for routing
    breakpoints := make([]int, len(shards))
    for i, s := range shards {
        breakpoints[i] = s.RangeStart
    }
    return &RangeShardRouter{shards: shards, breakpoints: breakpoints}
}

func (r *RangeShardRouter) Route(key int) *Shard {
    // Binary search returns insertion point — subtract 1 to find owning shard
    // => example: key=15000, breakpoints=[1,10001,20001], search=2 -> shard index 1
    idx := sort.SearchInts(r.breakpoints, key+1) - 1
    if idx < 0 {
        return nil // key is below first shard's RangeStart
    }
    shard := r.shards[idx]
    // Verify key is within this shard's upper bound (handles gaps in ranges)
    if key > shard.RangeEnd {
        return nil // key falls in an unassigned gap
    }
    return shard
}

func (r *RangeShardRouter) Insert(key int, value map[string]string) error {
    shard := r.Route(key)
    if shard == nil {
        return fmt.Errorf("no shard configured for key=%d", key)
    }
    shard.Insert(key, value)
    return nil
}

func (r *RangeShardRouter) Get(key int) (map[string]string, bool) {
    shard := r.Route(key)
    if shard == nil {
        return nil, false
    }
    return shard.Get(key)
}

func (r *RangeShardRouter) CrossShardRangeScan(start, end int) []map[string]string {
    // Cross-shard range scan: must query every shard whose range overlaps [start, end]
    // => this is why range sharding is good for range queries within a shard
    // => but cross-shard scans still require fan-out to multiple shards
    var results []map[string]string
    for _, shard := range r.shards {
        if shard.RangeStart <= end && shard.RangeEnd >= start {
            results = append(results, shard.RangeScan(start, end)...)
        }
    }
    return results
}

func main() {
    // Configure three shards covering user IDs 1 to 30000
    shards := []*Shard{
        NewShard(0, 1, 10000),
        NewShard(1, 10001, 20000),
        NewShard(2, 20001, 30000),
    }
    router := NewRangeShardRouter(shards)
    // => router.breakpoints = [1, 10001, 20001]

    // Insert records — router sends each to the correct shard
    router.Insert(500, map[string]string{"name": "Alice", "plan": "free"})
    // => [Shard-0] INSERT key=500 range=[1,10000]
    router.Insert(15000, map[string]string{"name": "Bob", "plan": "pro"})
    // => [Shard-1] INSERT key=15000 range=[10001,20000]
    router.Insert(25000, map[string]string{"name": "Charlie", "plan": "enterprise"})
    // => [Shard-2] INSERT key=25000 range=[20001,30000]

    // Point lookup — O(log n) routing via binary search, then O(1) store lookup
    record, _ := router.Get(15000)
    fmt.Printf("GET 15000: %v\n", record)
    // => GET 15000: map[name:Bob plan:pro]

    // Range scan crossing shard boundary — fan-out to shards 0 and 1
    results := router.CrossShardRangeScan(9000, 11000)
    fmt.Printf("Range scan [9000, 11000]: %d records\n", len(results))
    // => Range scan [9000, 11000]: 2 records (Alice from Shard-0, Bob from Shard-1)

    // Hotspot demonstration: all new users cluster at the top of the range
    // => shard-2 receives all traffic while shard-0 and shard-1 sit idle
    // => solution: hash sharding (Example 34) distributes writes uniformly
}