Advanced

// CAP theorem demonstration: simulating CP vs AP behavior during partition
package main

import (
    "fmt"
    "errors"
    "sync"
)

type CPDatabase struct {
    data        map[string]string // => primary data store
    partitioned bool              // => partition flag; true means split-brain risk
    mu          sync.Mutex
}

func NewCPDatabase() *CPDatabase {
    return &CPDatabase{data: make(map[string]string)}
}

func (db *CPDatabase) Write(key, value string) bool {
    // => Writes always succeed on primary node
    db.mu.Lock()
    defer db.mu.Unlock()
    db.data[key] = value
    return true // => write confirmed
}

func (db *CPDatabase) Read(key string) (string, error) {
    // => CP: refuses reads during partition to prevent stale data
    if db.partitioned {
        return "", errors.New("Service unavailable: partition detected")
        // => raises instead of returning possibly stale value
    }
    db.mu.Lock()
    defer db.mu.Unlock()
    return db.data[key], nil // => returns current consistent value
}

type APDatabase struct {
    data        map[string]string // => local data, may diverge from other nodes
    partitioned bool
}

func NewAPDatabase() *APDatabase {
    return &APDatabase{data: make(map[string]string)}
}

func (db *APDatabase) Write(key, value string) bool {
    db.data[key] = value // => writes locally even during partition
    return true          // => always acknowledges write
}

func (db *APDatabase) Read(key string) string {
    // => AP: always responds, even if value is stale
    return db.data[key]
    // => caller cannot know if value is the most recent globally
}

func main() {
    // Demonstrate the tradeoff
    cpDB := NewCPDatabase() // => CP database instance
    apDB := NewAPDatabase() // => AP database instance

    cpDB.Write("user:1", "alice") // => stored: {"user:1": "alice"}
    apDB.Write("user:1", "alice") // => stored: {"user:1": "alice"}

    // Simulate partition
    cpDB.partitioned = true // => partition starts
    apDB.partitioned = true // => partition starts

    _, err := cpDB.Read("user:1") // => returns error (CP refuses)
    if err != nil {
        fmt.Printf("CP during partition: %v\n", err)
        // => Output: CP during partition: Service unavailable: partition detected
    }

    val := apDB.Read("user:1") // => returns "alice" (possibly stale)
    fmt.Printf("AP during partition: %s\n", val)
    // => Output: AP during partition: alice
}

# CAP theorem demonstration: simulating CP vs AP behavior during partition
import threading
import time

class CPDatabase:
    # CP system: consistency over availability
    # During a partition, refuses to serve stale reads
    def __init__(self):
        self.data = {}          # => primary data store
        self.partitioned = False # => partition flag; True means split-brain risk
        self.lock = threading.Lock()

    def write(self, key, value):
        # => Writes always succeed on primary node
        with self.lock:
            self.data[key] = value
            return True         # => write confirmed

    def read(self, key):
        # => CP: refuses reads during partition to prevent stale data
        if self.partitioned:
            raise Exception("Service unavailable: partition detected")
            # => raises instead of returning possibly stale value
        with self.lock:
            return self.data.get(key)   # => returns current consistent value


class APDatabase:
    # AP system: availability over consistency
    # During a partition, returns last-known value (possibly stale)
    def __init__(self):
        self.data = {}          # => local data, may diverge from other nodes
        self.partitioned = False

    def write(self, key, value):
        self.data[key] = value  # => writes locally even during partition
        return True             # => always acknowledges write

    def read(self, key):
        # => AP: always responds, even if value is stale
        return self.data.get(key)
        # => caller cannot know if value is the most recent globally


# Demonstrate the tradeoff
cp_db = CPDatabase()            # => CP database instance
ap_db = APDatabase()            # => AP database instance

cp_db.write("user:1", "alice")  # => stored: {"user:1": "alice"}
ap_db.write("user:1", "alice")  # => stored: {"user:1": "alice"}

# Simulate partition
cp_db.partitioned = True        # => partition starts
ap_db.partitioned = True        # => partition starts

try:
    val = cp_db.read("user:1")  # => raises Exception (CP refuses)
except Exception as e:
    print(f"CP during partition: {e}")
    # => Output: CP during partition: Service unavailable: partition detected

val = ap_db.read("user:1")      # => returns "alice" (possibly stale)
print(f"AP during partition: {val}")
# => Output: AP during partition: alice

// PACELC tradeoff simulation: replication consistency vs latency
package main

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

type ReplicationStrategy struct {
    nodes []map[string]string // => 3-node cluster; node[0] is primary
}

func NewReplicationStrategy(n int) *ReplicationStrategy {
    nodes := make([]map[string]string, n)
    for i := range nodes {
        nodes[i] = make(map[string]string)
    }
    return &ReplicationStrategy{nodes: nodes}
}

func (rs *ReplicationStrategy) WriteSynchronous(key, value string) map[string]interface{} {
    // PC/EC path: wait for all replicas before acknowledging.
    start := time.Now()
    // Simulate replication to all nodes with network delay
    for _, node := range rs.nodes {
        time.Sleep(10 * time.Millisecond) // => 10ms per replica round-trip
        node[key] = value
    }
    latencyMs := float64(time.Since(start).Milliseconds())
    // => All 3 nodes written: latency = ~30ms (3 x 10ms)
    return map[string]interface{}{
        "ack": true, "consistent": true, "latency_ms": latencyMs,
    }
}

func (rs *ReplicationStrategy) WriteAsynchronous(key, value string) map[string]interface{} {
    // PA/EL path: acknowledge after primary write, replicate async.
    start := time.Now()
    rs.nodes[0][key] = value // => write to primary only
    latencyMs := float64(time.Since(start).Microseconds()) / 1000.0
    // => Only primary written: latency = ~0ms
    // Async replication happens eventually (simulated)
    go func() {
        for _, node := range rs.nodes[1:] {
            time.Sleep(10 * time.Millisecond) // => background replication, caller unblocked
            node[key] = value
        }
    }() // => fire-and-forget replication
    return map[string]interface{}{
        "ack": true, "consistent": false, "latency_ms": latencyMs,
    }
}

func (rs *ReplicationStrategy) ReadQuorum(key string, quorum int) map[string]interface{} {
    // Read from majority quorum to detect inconsistency.
    values := make([]string, 0, quorum)
    for _, node := range rs.nodes[:quorum] {
        values = append(values, node[key])
        // => read from quorum nodes; majority rules
    }
    consistent := true
    for _, v := range values[1:] {
        if v != values[0] {
            consistent = false
            break
        }
    }
    // => consistent=true if all quorum nodes agree
    return map[string]interface{}{
        "value": values[0], "consistent": consistent,
    }
}

func main() {
    strategy := NewReplicationStrategy(3)

    syncResult := strategy.WriteSynchronous("order:42", "confirmed")
    // => {"ack": true, "consistent": true, "latency_ms": 30}
    fmt.Printf("Sync write: %v\n", syncResult)

    asyncResult := strategy.WriteAsynchronous("feed:99", "post_data")
    // => {"ack": true, "consistent": false, "latency_ms": 0.x}
    fmt.Printf("Async write: %v\n", asyncResult)
}

# PACELC tradeoff simulation: replication consistency vs latency
import time
import random

class ReplicationStrategy:
    """Demonstrates latency vs consistency tradeoff in replication."""

    def __init__(self, nodes=3):
        self.nodes = [{"data": {}} for _ in range(nodes)]
        # => 3-node cluster; node[0] is primary

    def write_synchronous(self, key, value):
        """PC/EC path: wait for all replicas before acknowledging."""
        start = time.time()
        # Simulate replication to all nodes with network delay
        for node in self.nodes:
            time.sleep(0.01)    # => 10ms per replica round-trip
            node["data"][key] = value
        latency_ms = (time.time() - start) * 1000
        # => All 3 nodes written: latency = ~30ms (3 x 10ms)
        return {"ack": True, "consistent": True, "latency_ms": round(latency_ms)}

    def write_asynchronous(self, key, value):
        """PA/EL path: acknowledge after primary write, replicate async."""
        start = time.time()
        self.nodes[0]["data"][key] = value  # => write to primary only
        latency_ms = (time.time() - start) * 1000
        # => Only primary written: latency = ~0ms
        # Async replication happens eventually (simulated)
        def replicate():
            for node in self.nodes[1:]:
                time.sleep(0.01)    # => background replication, caller unblocked
                node["data"][key] = value
        t = threading.Thread(target=replicate, daemon=True)
        import threading
        t = threading.Thread(target=replicate, daemon=True)
        t.start()               # => fire-and-forget replication
        return {"ack": True, "consistent": False, "latency_ms": round(latency_ms, 2)}

    def read_quorum(self, key, quorum=2):
        """Read from majority quorum to detect inconsistency."""
        values = []
        for node in self.nodes[:quorum]:
            values.append(node["data"].get(key))
            # => read from quorum nodes; majority rules
        consistent = len(set(str(v) for v in values)) == 1
        # => consistent=True if all quorum nodes agree
        return {"value": values[0], "consistent": consistent}

strategy = ReplicationStrategy()

sync_result = strategy.write_synchronous("order:42", "confirmed")
# => {"ack": True, "consistent": True, "latency_ms": 30}
print(f"Sync write: {sync_result}")

async_result = strategy.write_asynchronous("feed:99", "post_data")
# => {"ack": True, "consistent": False, "latency_ms": 0.x}
print(f"Async write: {async_result}")

// Raft leader election simulation (simplified)
package main

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

const (
    Follower  = "follower"
    Candidate = "candidate"
    Leader    = "leader"
)

type RaftNode struct {
    ID              int
    State           string
    CurrentTerm     int
    VotedFor        int    // => -1 means no vote
    VotesReceived   int
    ClusterSize     int
    ElectionTimeout time.Duration
}

func NewRaftNode(id, clusterSize int) *RaftNode {
    // Randomized timeout: 150-300ms (Raft spec range)
    timeout := time.Duration(150+rand.Intn(150)) * time.Millisecond
    // => random timeout prevents simultaneous elections (split votes)
    return &RaftNode{
        ID:              id,
        State:           Follower,  // => all nodes start as followers
        CurrentTerm:     0,         // => monotonically increasing election term
        VotedFor:        -1,        // => no vote yet
        ClusterSize:     clusterSize, // => total nodes; majority = size/2 + 1
        ElectionTimeout: timeout,
    }
}

func (n *RaftNode) StartElection() int {
    // Transition to candidate and request votes.
    n.State = Candidate        // => become candidate
    n.CurrentTerm++            // => increment term (new election epoch)
    n.VotedFor = n.ID          // => vote for self
    n.VotesReceived = 1        // => count self-vote
    fmt.Printf("Node %d: starting election for term %d\n", n.ID, n.CurrentTerm)
    return n.CurrentTerm       // => return term for RequestVote RPCs
}

func (n *RaftNode) ReceiveVote(voterID, term int) {
    // Process an incoming vote grant.
    if term == n.CurrentTerm && n.State == Candidate {
        n.VotesReceived++      // => tally the vote
        majority := n.ClusterSize/2 + 1
        // => majority required: 3 of 5, 2 of 3, etc.
        if n.VotesReceived >= majority {
            n.State = Leader   // => won election!
            fmt.Printf("Node %d: elected LEADER for term %d\n", n.ID, n.CurrentTerm)
            fmt.Printf("  Votes: %d/%d\n", n.VotesReceived, n.ClusterSize)
        }
    }
}

func (n *RaftNode) GrantVote(candidateID, candidateTerm int) bool {
    // Decide whether to grant vote to candidate.
    if candidateTerm > n.CurrentTerm {
        n.CurrentTerm = candidateTerm // => update to higher term
        n.State = Follower            // => step down if was leader/candidate
        n.VotedFor = -1               // => reset vote for new term
    }
    if n.VotedFor == -1 || n.VotedFor == candidateID {
        n.VotedFor = candidateID      // => grant vote
        return true                   // => true = vote granted
    }
    return false                      // => false = already voted this term
}

func main() {
    // Simulate 5-node cluster election
    nodes := make([]*RaftNode, 5)
    for i := range nodes {
        nodes[i] = NewRaftNode(i, 5)
    }
    // => [node0, node1, node2, node3, node4], all followers, term=0

    // Node 0's timeout fires first (smallest random timeout)
    term := nodes[0].StartElection()
    // => node0 becomes candidate, term=1, votes=1

    // Other nodes grant votes
    for _, n := range nodes[1:] {
        granted := n.GrantVote(0, term) // => each follower evaluates vote request
        if granted {
            nodes[0].ReceiveVote(n.ID, term)
            // => node0 accumulates votes; becomes leader at 3/5
        }
    }
    // Output shows node0 becoming leader when majority (3) grants vote
}

# Raft leader election simulation (simplified)
import random
import time

class RaftNode:
    FOLLOWER = "follower"
    CANDIDATE = "candidate"
    LEADER = "leader"

    def __init__(self, node_id, cluster_size):
        self.id = node_id                   # => unique node identifier
        self.state = self.FOLLOWER          # => all nodes start as followers
        self.current_term = 0               # => monotonically increasing election term
        self.voted_for = None               # => candidate voted for in current term
        self.votes_received = 0             # => votes received as candidate
        self.cluster_size = cluster_size    # => total nodes; majority = size//2 + 1
        # Randomized timeout: 150-300ms (Raft spec range)
        self.election_timeout = random.uniform(0.15, 0.30)
        # => random timeout prevents simultaneous elections (split votes)

    def start_election(self):
        """Transition to candidate and request votes."""
        self.state = self.CANDIDATE         # => become candidate
        self.current_term += 1              # => increment term (new election epoch)
        self.voted_for = self.id            # => vote for self
        self.votes_received = 1             # => count self-vote
        print(f"Node {self.id}: starting election for term {self.current_term}")
        return self.current_term            # => return term for RequestVote RPCs

    def receive_vote(self, voter_id, term):
        """Process an incoming vote grant."""
        if term == self.current_term and self.state == self.CANDIDATE:
            self.votes_received += 1        # => tally the vote
            majority = self.cluster_size // 2 + 1
            # => majority required: 3 of 5, 2 of 3, etc.
            if self.votes_received >= majority:
                self.state = self.LEADER    # => won election!
                print(f"Node {self.id}: elected LEADER for term {self.current_term}")
                print(f"  Votes: {self.votes_received}/{self.cluster_size}")

    def grant_vote(self, candidate_id, candidate_term):
        """Decide whether to grant vote to candidate."""
        if candidate_term > self.current_term:
            self.current_term = candidate_term  # => update to higher term
            self.state = self.FOLLOWER          # => step down if was leader/candidate
            self.voted_for = None               # => reset vote for new term
        if (self.voted_for is None or self.voted_for == candidate_id):
            self.voted_for = candidate_id       # => grant vote
            return True                         # => True = vote granted
        return False                            # => False = already voted this term

# Simulate 5-node cluster election
nodes = [RaftNode(i, 5) for i in range(5)]
# => [node0, node1, node2, node3, node4], all followers, term=0

# Node 0's timeout fires first (smallest random timeout)
term = nodes[0].start_election()
# => node0 becomes candidate, term=1, votes=1

# Other nodes grant votes
for n in nodes[1:]:
    granted = n.grant_vote(0, term)     # => each follower evaluates vote request
    if granted:
        nodes[0].receive_vote(n.id, term)
        # => node0 accumulates votes; becomes leader at 3/5

# Output shows node0 becoming leader when majority (3) grants vote

// Paxos single-decree consensus simulation
package main

import "fmt"

type Acceptor struct {
    ID            int
    PromisedID    int         // => highest proposal number promised
    AcceptedID    int         // => proposal number of last accepted value
    AcceptedValue string      // => last accepted value (if any)
}

func NewAcceptor(id int) *Acceptor {
    return &Acceptor{ID: id, PromisedID: -1, AcceptedID: -1}
}

type PrepareResponse struct {
    Promise       bool
    AcceptedID    int
    AcceptedValue string
}

func (a *Acceptor) Prepare(proposalID int) PrepareResponse {
    // Phase 1: respond to Prepare(n) from proposer.
    if proposalID > a.PromisedID {
        a.PromisedID = proposalID
        // => promise to reject proposals with id < proposalID
        return PrepareResponse{
            Promise:       true,
            AcceptedID:    a.AcceptedID,    // => tell proposer what we accepted before
            AcceptedValue: a.AcceptedValue,
            // => proposer must use this value if non-empty (safety constraint)
        }
    }
    return PrepareResponse{Promise: false} // => already promised to a higher proposal
}

type AcceptResponse struct {
    Accepted bool
    Value    string
}

func (a *Acceptor) Accept(proposalID int, value string) AcceptResponse {
    // Phase 2: respond to Accept(n, v) from proposer.
    if proposalID >= a.PromisedID {
        // => only accept if we haven't promised something higher
        a.AcceptedID = proposalID
        a.AcceptedValue = value     // => record the accepted value
        a.PromisedID = proposalID   // => update promise to this id
        return AcceptResponse{Accepted: true, Value: value}
    }
    return AcceptResponse{Accepted: false} // => rejected: already promised higher
}

type Proposer struct {
    ID              int
    Acceptors       []*Acceptor // => list of Acceptor objects
    ProposalCounter int         // => local counter for unique proposal IDs
}

func NewProposer(id int, acceptors []*Acceptor) *Proposer {
    return &Proposer{ID: id, Acceptors: acceptors}
}

func (p *Proposer) Propose(desiredValue string) (string, bool) {
    // Run full Paxos protocol to get consensus on a value.
    p.ProposalCounter++
    // => unique proposal_id: combine proposer_id and counter
    proposalID := p.ID*100 + p.ProposalCounter

    // --- Phase 1: Prepare ---
    var promises []PrepareResponse
    for _, acceptor := range p.Acceptors {
        response := acceptor.Prepare(proposalID)
        // => send Prepare(proposalID) to each acceptor
        if response.Promise {
            promises = append(promises, response) // => collect promises
        }
    }

    majority := len(p.Acceptors)/2 + 1
    if len(promises) < majority {
        return "", false // => failed to get majority promises; retry needed
    }

    // Safety: if any acceptor already accepted a value, use highest-id one
    valueToPropose := desiredValue
    highestAcceptedID := -1
    for _, pr := range promises {
        if pr.AcceptedValue != "" && pr.AcceptedID > highestAcceptedID {
            // => must preserve previously chosen value (Paxos safety rule)
            highestAcceptedID = pr.AcceptedID
            valueToPropose = pr.AcceptedValue
            // => cannot use desiredValue; must preserve consensus
        }
    }

    // --- Phase 2: Accept ---
    acceptedCount := 0
    for _, acceptor := range p.Acceptors {
        response := acceptor.Accept(proposalID, valueToPropose)
        if response.Accepted {
            acceptedCount++ // => tally acceptances
        }
    }

    if acceptedCount >= majority {
        return valueToPropose, true // => consensus reached!
    }
    return "", false               // => failed; retry with higher id
}

func main() {
    // 3-acceptor Paxos run
    acceptors := []*Acceptor{NewAcceptor(0), NewAcceptor(1), NewAcceptor(2)}
    // => a0, a1, a2
    proposer := NewProposer(1, acceptors)

    result, ok := proposer.Propose("commit_transaction_42")
    // => Phase 1: all 3 promise; no prior accepted values
    // => Phase 2: all 3 accept "commit_transaction_42"
    // => result = "commit_transaction_42" (consensus achieved)
    if ok {
        fmt.Printf("Consensus: %s\n", result)
    }
}

# Paxos single-decree consensus simulation
class Acceptor:
    def __init__(self, acceptor_id):
        self.id = acceptor_id
        self.promised_id = -1       # => highest proposal number promised
        self.accepted_id = -1       # => proposal number of last accepted value
        self.accepted_value = None  # => last accepted value (if any)

    def prepare(self, proposal_id):
        """Phase 1: respond to Prepare(n) from proposer."""
        if proposal_id > self.promised_id:
            self.promised_id = proposal_id
            # => promise to reject proposals with id < proposal_id
            return {
                "promise": True,
                "accepted_id": self.accepted_id,    # => tell proposer what we accepted before
                "accepted_value": self.accepted_value
                # => proposer must use this value if non-None (safety constraint)
            }
        return {"promise": False}   # => already promised to a higher proposal

    def accept(self, proposal_id, value):
        """Phase 2: respond to Accept(n, v) from proposer."""
        if proposal_id >= self.promised_id:
            # => only accept if we haven't promised something higher
            self.accepted_id = proposal_id
            self.accepted_value = value         # => record the accepted value
            self.promised_id = proposal_id      # => update promise to this id
            return {"accepted": True, "value": value}
        return {"accepted": False}              # => rejected: already promised higher


class Proposer:
    def __init__(self, proposer_id, acceptors):
        self.id = proposer_id
        self.acceptors = acceptors          # => list of Acceptor objects
        self.proposal_counter = 0           # => local counter for unique proposal IDs

    def propose(self, desired_value):
        """Run full Paxos protocol to get consensus on a value."""
        self.proposal_counter += 1
        # => unique proposal_id: combine proposer_id and counter
        proposal_id = self.id * 100 + self.proposal_counter

        # --- Phase 1: Prepare ---
        promises = []
        for acceptor in self.acceptors:
            response = acceptor.prepare(proposal_id)
            # => send Prepare(proposal_id) to each acceptor
            if response["promise"]:
                promises.append(response)   # => collect promises

        majority = len(self.acceptors) // 2 + 1
        if len(promises) < majority:
            return None             # => failed to get majority promises; retry needed

        # Safety: if any acceptor already accepted a value, use highest-id one
        prior = [p for p in promises if p["accepted_value"] is not None]
        if prior:
            # => must preserve previously chosen value (Paxos safety rule)
            chosen_prior = max(prior, key=lambda p: p["accepted_id"])
            value_to_propose = chosen_prior["accepted_value"]
            # => cannot use desired_value; must preserve consensus
        else:
            value_to_propose = desired_value    # => free to propose our value

        # --- Phase 2: Accept ---
        accepted_count = 0
        for acceptor in self.acceptors:
            response = acceptor.accept(proposal_id, value_to_propose)
            if response["accepted"]:
                accepted_count += 1             # => tally acceptances

        if accepted_count >= majority:
            return value_to_propose             # => consensus reached!
        return None                             # => failed; retry with higher id

# 3-acceptor Paxos run
acceptors = [Acceptor(i) for i in range(3)]    # => a0, a1, a2
proposer = Proposer(proposer_id=1, acceptors=acceptors)

result = proposer.propose("commit_transaction_42")
# => Phase 1: all 3 promise; no prior accepted values
# => Phase 2: all 3 accept "commit_transaction_42"
# => result = "commit_transaction_42" (consensus achieved)
print(f"Consensus: {result}")

// Event sourcing: order aggregate with immutable event log
package main

import "fmt"

type Event struct {
    EventType   string            // => "OrderPlaced", "ItemAdded", "OrderShipped", etc.
    AggregateID string            // => which aggregate this event belongs to
    Timestamp   string            // => ISO timestamp for ordering
    Data        map[string]interface{} // => event payload; immutable after append
}

type EventStore struct {
    log []Event // => the immutable log
}

func (es *EventStore) Append(event Event) {
    es.log = append(es.log, event) // => append only; no updates
    fmt.Printf("Appended: %s @ %s\n", event.EventType, event.Timestamp)
}

func (es *EventStore) GetEvents(aggregateID string) []Event {
    var events []Event
    for _, e := range es.log {
        if e.AggregateID == aggregateID {
            events = append(events, e)
        }
    }
    return events // => filter log for one aggregate's history
}

type OrderState struct {
    Items    []string
    Status   string
    Total    float64
    Customer string
}

type OrderAggregate struct {
    OrderID string
    Store   *EventStore
    State   OrderState // => state derived from events, not stored directly
}

func NewOrderAggregate(orderID string, store *EventStore) *OrderAggregate {
    oa := &OrderAggregate{OrderID: orderID, Store: store}
    oa.State = oa.replay() // => state derived from events, not stored directly
    return oa
}

func (oa *OrderAggregate) replay() OrderState {
    // Rebuild state from event log — the core ES mechanism.
    state := OrderState{Status: "new"}
    for _, event := range oa.Store.GetEvents(oa.OrderID) {
        // => apply each event in order to evolve state
        switch event.EventType {
        case "OrderPlaced":
            state.Status = "placed"
            state.Customer = event.Data["customer"].(string)
        case "ItemAdded":
            state.Items = append(state.Items, event.Data["item"].(string))
            state.Total += event.Data["price"].(float64)
            // => total accumulates from ItemAdded events
        case "OrderShipped":
            state.Status = "shipped"
        }
    }
    return state // => current state as of all replayed events
}

func (oa *OrderAggregate) PlaceOrder(customer string) {
    event := Event{
        EventType:   "OrderPlaced",
        AggregateID: oa.OrderID,
        Timestamp:   "2026-03-20T10:00:00Z",
        Data:        map[string]interface{}{"customer": customer},
    }
    oa.Store.Append(event)  // => emit event; do NOT write state directly
    oa.State = oa.replay()  // => update in-memory state
}

func (oa *OrderAggregate) AddItem(item string, price float64) {
    event := Event{
        EventType:   "ItemAdded",
        AggregateID: oa.OrderID,
        Timestamp:   "2026-03-20T10:01:00Z",
        Data:        map[string]interface{}{"item": item, "price": price},
    }
    oa.Store.Append(event)
    oa.State = oa.replay()
}

func main() {
    // Run event sourcing demo
    store := &EventStore{}
    order := NewOrderAggregate("order-42", store)

    order.PlaceOrder("alice")        // => appends OrderPlaced event
    order.AddItem("laptop", 999.99)  // => appends ItemAdded event
    order.AddItem("mouse", 29.99)    // => appends ItemAdded event

    fmt.Printf("Current state: %+v\n", order.State)
    // => {Items:[laptop mouse] Status:placed Total:1029.98 Customer:alice}

    // Temporal query: reconstruct state after first event only
    pastEvents := store.GetEvents("order-42")[:1]
    // => only OrderPlaced event; state would be {Items:[], Status:placed, ...}
    _ = pastEvents
}

# Event sourcing: order aggregate with immutable event log
from dataclasses import dataclass, field
from typing import List, Optional
from datetime import datetime

@dataclass
class Event:
    event_type: str         # => "OrderPlaced", "ItemAdded", "OrderShipped", etc.
    aggregate_id: str       # => which aggregate this event belongs to
    timestamp: str          # => ISO timestamp for ordering
    data: dict              # => event payload; immutable after append

class EventStore:
    """Append-only event log — never update or delete events."""

    def __init__(self):
        self._log: List[Event] = []     # => the immutable log

    def append(self, event: Event):
        self._log.append(event)         # => append only; no updates
        print(f"Appended: {event.event_type} @ {event.timestamp}")

    def get_events(self, aggregate_id: str) -> List[Event]:
        return [e for e in self._log if e.aggregate_id == aggregate_id]
        # => filter log for one aggregate's history

class OrderAggregate:
    """Reconstitutes current state by replaying events."""

    def __init__(self, order_id: str, store: EventStore):
        self.order_id = order_id
        self.store = store
        # Replay all past events to restore current state
        self._state = self._replay()    # => state derived from events, not stored directly

    def _replay(self) -> dict:
        """Rebuild state from event log — the core ES mechanism."""
        state = {"items": [], "status": "new", "total": 0}
        for event in self.store.get_events(self.order_id):
            # => apply each event in order to evolve state
            if event.event_type == "OrderPlaced":
                state["status"] = "placed"
                state["customer"] = event.data["customer"]
            elif event.event_type == "ItemAdded":
                state["items"].append(event.data["item"])
                state["total"] += event.data["price"]
                # => total accumulates from ItemAdded events
            elif event.event_type == "OrderShipped":
                state["status"] = "shipped"
                state["tracking"] = event.data["tracking_number"]
        return state    # => current state as of all replayed events

    def place_order(self, customer: str):
        event = Event("OrderPlaced", self.order_id, "2026-03-20T10:00:00Z",
                      {"customer": customer})
        self.store.append(event)        # => emit event; do NOT write state directly
        self._state = self._replay()    # => update in-memory state

    def add_item(self, item: str, price: float):
        event = Event("ItemAdded", self.order_id, "2026-03-20T10:01:00Z",
                      {"item": item, "price": price})
        self.store.append(event)
        self._state = self._replay()

    @property
    def state(self) -> dict:
        return self._state              # => always derived from events

# Run event sourcing demo
store = EventStore()
order = OrderAggregate("order-42", store)

order.place_order("alice")              # => appends OrderPlaced event
order.add_item("laptop", 999.99)        # => appends ItemAdded event
order.add_item("mouse", 29.99)          # => appends ItemAdded event

print(f"Current state: {order.state}")
# => {"items": ["laptop","mouse"], "status":"placed", "total":1029.98, "customer":"alice"}

# Temporal query: reconstruct state after first event only
past_state_events = store.get_events("order-42")[:1]
# => only OrderPlaced event; state would be {"items":[], "status":"placed", ...}

// CQRS: separate command and query handlers with independent models
package main

import (
    "fmt"
)

// --- Write Side: Command Model ---
type Product struct {
    ProductID string
    Name      string
    Price     float64
    Stock     int
}

type ProductCommandHandler struct {
    products map[string]*Product // => write-side store
    events   []map[string]interface{} // => event log for projection
}

func NewProductCommandHandler() *ProductCommandHandler {
    return &ProductCommandHandler{
        products: make(map[string]*Product),
    }
}

func (h *ProductCommandHandler) CreateProduct(productID, name string, price float64, stock int) error {
    if _, exists := h.products[productID]; exists {
        return fmt.Errorf("Product %s already exists", productID)
        // => command handler enforces uniqueness invariant
    }
    h.products[productID] = &Product{productID, name, price, stock}
    // Emit event for read-side projection to consume
    h.events = append(h.events, map[string]interface{}{
        "type": "ProductCreated", "id": productID,
        "name": name, "price": price, "stock": stock,
    })
    // => event published; query side will update eventually
    return nil
}

func (h *ProductCommandHandler) UpdateStock(productID string, delta int) error {
    product := h.products[productID]
    newStock := product.Stock + delta
    if newStock < 0 {
        return fmt.Errorf("Stock cannot go negative")
        // => command model enforces non-negative stock invariant
    }
    product.Stock = newStock
    h.events = append(h.events, map[string]interface{}{
        "type": "StockUpdated", "id": productID, "new_stock": newStock,
    })
    return nil
}

// --- Read Side: Query Model (Projection) ---
type CatalogItem struct {
    ID           string
    Name         string
    Price        float64
    Stock        int
    Available    bool   // => computed field for queries
    DisplayPrice string // => pre-formatted for API responses
}

type ProductQueryHandler struct {
    catalog []CatalogItem // => flat list for O(1) search
}

func (q *ProductQueryHandler) ApplyEvent(event map[string]interface{}) {
    // Update projection when event arrives from write side.
    switch event["type"] {
    case "ProductCreated":
        stock := event["stock"].(int)
        price := event["price"].(float64)
        q.catalog = append(q.catalog, CatalogItem{
            ID:           event["id"].(string),
            Name:         event["name"].(string),
            Price:        price,
            Stock:        stock,
            Available:    stock > 0,                    // => computed field for queries
            DisplayPrice: fmt.Sprintf("$%.2f", price),  // => pre-formatted; avoids repeated computation
        })
    case "StockUpdated":
        id := event["id"].(string)
        newStock := event["new_stock"].(int)
        for i := range q.catalog {
            if q.catalog[i].ID == id {
                q.catalog[i].Stock = newStock
                q.catalog[i].Available = newStock > 0
                // => projection updated asynchronously; may lag briefly
            }
        }
    }
}

func (q *ProductQueryHandler) GetAvailableProducts() []CatalogItem {
    // Read query: fast scan of denormalized projection.
    var result []CatalogItem
    for _, p := range q.catalog {
        if p.Available {
            result = append(result, p)
        }
    }
    return result
    // => no joins, no computation; just filter pre-built projection
}

func main() {
    // Wire up CQRS
    commands := NewProductCommandHandler()
    queries := &ProductQueryHandler{}

    commands.CreateProduct("p1", "Laptop", 999.99, 10)
    commands.CreateProduct("p2", "Mouse", 29.99, 0)  // => stock=0, not available
    commands.CreateProduct("p3", "Keyboard", 79.99, 5)

    // Sync events to query side (in production: Kafka/SNS/outbox pattern)
    for _, event := range commands.events {
        queries.ApplyEvent(event) // => projection updated from events
    }

    available := queries.GetAvailableProducts()
    // => [Laptop, Keyboard]; p2 excluded because stock=0
    names := make([]string, len(available))
    for i, p := range available {
        names[i] = p.Name
    }
    fmt.Printf("Available: %v\n", names)
    // => Output: Available: [Laptop Keyboard]
}