মাল্টি-রিজিয়ন আর্কিটেকচার

package region

// Region-aware request router with health tracking and fenced failover.

import (
	"fmt"
	"log"
	"sort"
	"sync"
	"time"
)

type RegionID string
type Residency string

const (
	Baghdad   RegionID = "baghdad"
	Cordoba   RegionID = "cordoba"
	Samarkand RegionID = "samarkand"

	EUOnly Residency = "eu-only"
	MEOnly Residency = "me-only"
	AnyRes Residency = "any"

	failureThreshold  = 3
	maxStaleReadLagMs = 10_000
)

var residencyAllowed = map[Residency][]RegionID{
	EUOnly: {Cordoba},
	MEOnly: {Baghdad},
	AnyRes: {Baghdad, Cordoba, Samarkand},
}

func allowedIn(res Residency, id RegionID) bool {
	for _, r := range residencyAllowed[res] {
		if r == id { return true }
	}
	return false
}

type RegionHealth struct {
	ID                  RegionID
	Healthy             bool
	ConsecutiveFailures int
	LastProbeAt         time.Time
	RTTMs               int64
	ReplicationLagMs    int64
}

type TenantPlacement struct {
	TenantID   string
	HomeRegion RegionID
	Residency  Residency
	Epoch      int64 // bumped by the control plane on every ownership change
}

type RouteMode string

const (
	LocalRead     RouteMode = "local-read"
	HomeRead      RouteMode = "home-read"
	HomeWrite     RouteMode = "home-write"
	FailoverWrite RouteMode = "failover-write"
)

type RouteDecision struct {
	Target           RegionID
	Mode             RouteMode
	Epoch            int64
	StaleReadAllowed bool
	Reason           string
}

// --- Registry -------------------------------------------------------------

type RegionRegistry struct {
	mu     sync.RWMutex
	health map[RegionID]*RegionHealth
}

func NewRegionRegistry(ids ...RegionID) *RegionRegistry {
	r := &RegionRegistry{health: make(map[RegionID]*RegionHealth, len(ids))}
	for _, id := range ids {
		r.health[id] = &RegionHealth{ID: id, Healthy: true, LastProbeAt: time.Now()}
	}
	return r
}

func (r *RegionRegistry) RecordProbe(id RegionID, ok bool, rttMs, lagMs int64) {
	r.mu.Lock()
	defer r.mu.Unlock()

	h, exists := r.health[id]
	if !exists { return }
	h.LastProbeAt, h.RTTMs, h.ReplicationLagMs = time.Now(), rttMs, lagMs

	if ok {
		h.ConsecutiveFailures, h.Healthy = 0, true
		return
	}
	// A single failed probe is noise; sustained failure is a signal.
	h.ConsecutiveFailures++
	if h.ConsecutiveFailures >= failureThreshold { h.Healthy = false }
}

func (r *RegionRegistry) Get(id RegionID) (RegionHealth, bool) {
	r.mu.RLock()
	defer r.mu.RUnlock()
	h, ok := r.health[id]
	if !ok { return RegionHealth{}, false }
	return *h, true
}

func (r *RegionRegistry) IsHealthy(id RegionID) bool {
	h, ok := r.Get(id)
	return ok && h.Healthy
}

// --- Placement store ------------------------------------------------------

type PlacementStore struct {
	mu         sync.RWMutex
	placements map[string]TenantPlacement
}

func NewPlacementStore() *PlacementStore {
	return &PlacementStore{placements: make(map[string]TenantPlacement)}
}

func (s *PlacementStore) Put(p TenantPlacement) error {
	s.mu.Lock()
	defer s.mu.Unlock()

	// Reject stale ownership claims — this is the fencing check.
	if existing, ok := s.placements[p.TenantID]; ok && p.Epoch <= existing.Epoch {
		return fmt.Errorf("stale placement for %s: epoch %d <= current %d", p.TenantID, p.Epoch, existing.Epoch)
	}
	if !allowedIn(p.Residency, p.HomeRegion) {
		return fmt.Errorf("residency violation: %s not allowed for %s", p.HomeRegion, p.Residency)
	}
	s.placements[p.TenantID] = p
	return nil
}

func (s *PlacementStore) Get(tenantID string) (TenantPlacement, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	p, ok := s.placements[tenantID]
	return p, ok
}

// --- Router ---------------------------------------------------------------

type RegionRouter struct {
	local      RegionID
	registry   *RegionRegistry
	placements *PlacementStore
}

func NewRegionRouter(local RegionID, reg *RegionRegistry, pl *PlacementStore) *RegionRouter {
	return &RegionRouter{local: local, registry: reg, placements: pl}
}

func (rr *RegionRouter) Route(tenantID string, write, requireFresh bool) (RouteDecision, error) {
	p, ok := rr.placements.Get(tenantID)
	if !ok { return RouteDecision{}, fmt.Errorf("no placement registered for tenant %s", tenantID) }
	if !write { return rr.routeRead(p, requireFresh) }

	if rr.registry.IsHealthy(p.HomeRegion) {
		return RouteDecision{Target: p.HomeRegion, Mode: HomeWrite, Epoch: p.Epoch, Reason: "home region healthy"}, nil
	}

	target, ok := rr.pickFailoverTarget(p)
	if !ok {
		return RouteDecision{}, fmt.Errorf("no failover target for %s: residency %s pins it to %s", tenantID, p.Residency, p.HomeRegion)
	}
	return RouteDecision{
		Target: target, Mode: FailoverWrite, Epoch: p.Epoch,
		Reason: fmt.Sprintf("home region %s unhealthy, failing over", p.HomeRegion),
	}, nil
}

func (rr *RegionRouter) routeRead(p TenantPlacement, requireFresh bool) (RouteDecision, error) {
	local, found := rr.registry.Get(rr.local)
	localUsable := found && local.Healthy &&
		local.ReplicationLagMs <= maxStaleReadLagMs &&
		allowedIn(p.Residency, rr.local)

	if !requireFresh && localUsable {
		return RouteDecision{
			Target: rr.local, Mode: LocalRead, Epoch: p.Epoch, StaleReadAllowed: true,
			Reason: fmt.Sprintf("local replica within lag budget (%dms)", local.ReplicationLagMs),
		}, nil
	}

	if rr.registry.IsHealthy(p.HomeRegion) {
		reason := "local replica too stale"
		if requireFresh { reason = "read-your-writes required" }
		return RouteDecision{Target: p.HomeRegion, Mode: HomeRead, Epoch: p.Epoch, Reason: reason}, nil
	}

	// Home is down and the caller wants fresh data: degrade explicitly
	// rather than silently serving unknown-age data.
	if localUsable {
		return RouteDecision{
			Target: rr.local, Mode: LocalRead, Epoch: p.Epoch, StaleReadAllowed: true,
			Reason: "home region down, serving degraded stale read",
		}, nil
	}
	return RouteDecision{}, fmt.Errorf("no readable region for tenant %s", p.TenantID)
}

func (rr *RegionRouter) pickFailoverTarget(p TenantPlacement) (RegionID, bool) {
	var candidates []RegionHealth
	for _, id := range residencyAllowed[p.Residency] {
		if id == p.HomeRegion { continue }
		if h, ok := rr.registry.Get(id); ok && h.Healthy { candidates = append(candidates, h) }
	}
	if len(candidates) == 0 { return "", false }
	// Prefer the freshest replica: least data loss on promotion.
	sort.Slice(candidates, func(i, j int) bool {
		return candidates[i].ReplicationLagMs < candidates[j].ReplicationLagMs
	})
	return candidates[0].ID, true
}

// --- Control plane: quorum-gated promotion --------------------------------

type PromotionVote struct {
	Voter            RegionID
	AgreesHomeIsDown bool
	ObservedLagMs    int64
}

type PromotionResult struct {
	Promoted            bool
	NewHome             RegionID
	NewEpoch            int64
	EstimatedDataLossMs int64
	Reason              string
}

type FailoverCoordinator struct {
	registry    *RegionRegistry
	placements  *PlacementStore
	totalVoters int
}

func NewFailoverCoordinator(reg *RegionRegistry, pl *PlacementStore, voters int) *FailoverCoordinator {
	return &FailoverCoordinator{registry: reg, placements: pl, totalVoters: voters}
}

// Promote requires a strict majority of control-plane voters to agree that the
// home region is unreachable. Without this, a network partition lets both sides
// promote themselves — split-brain.
func (fc *FailoverCoordinator) Promote(tenantID string, votes []PromotionVote) PromotionResult {
	p, ok := fc.placements.Get(tenantID)
	if !ok { return PromotionResult{Reason: "unknown tenant"} }

	quorum := fc.totalVoters/2 + 1
	var agreeing, eligible []PromotionVote
	for _, v := range votes {
		if v.AgreesHomeIsDown { agreeing = append(agreeing, v) }
	}
	if len(agreeing) < quorum {
		return PromotionResult{Reason: fmt.Sprintf("no quorum: %d/%d, need %d", len(agreeing), fc.totalVoters, quorum)}
	}

	for _, v := range agreeing {
		if v.Voter == p.HomeRegion || !fc.registry.IsHealthy(v.Voter) { continue }
		if !allowedIn(p.Residency, v.Voter) { continue }
		eligible = append(eligible, v)
	}
	if len(eligible) == 0 { return PromotionResult{Reason: "quorum reached but no eligible target region"} }
	sort.Slice(eligible, func(i, j int) bool { return eligible[i].ObservedLagMs < eligible[j].ObservedLagMs })

	target, newEpoch := eligible[0], p.Epoch+1
	if err := fc.placements.Put(TenantPlacement{
		TenantID: tenantID, HomeRegion: target.Voter, Residency: p.Residency, Epoch: newEpoch,
	}); err != nil {
		return PromotionResult{Reason: err.Error()}
	}

	log.Printf("[FAILOVER] tenant=%s %s -> %s epoch=%d rpo=%dms",
		tenantID, p.HomeRegion, target.Voter, newEpoch, target.ObservedLagMs)

	return PromotionResult{
		Promoted: true, NewHome: target.Voter, NewEpoch: newEpoch,
		EstimatedDataLossMs: target.ObservedLagMs,
		Reason:              "quorum agreed, freshest eligible replica promoted",
	}
}

package replication

// Cross-region replicator with version vectors and pluggable conflict merge.

import (
	"fmt"
	"math"
	"strings"
	"sync"
	"time"
)

type RegionID string
type VersionVector map[RegionID]uint64

type Ordering int

const (
	Before Ordering = iota
	After
	Equal
	Concurrent
)

type Versioned struct {
	Key    string
	Value  any
	Vector VersionVector
	// WrittenAt is a wall clock, used only as an LWW tiebreak — never as
	// the ordering source.
	WrittenAt    time.Time
	OriginRegion RegionID
}

type ReplicationRecord struct {
	Entry       Versioned
	LogPosition uint64
	EmittedAt   time.Time
}

type MergeFn func(local, remote Versioned) any

type Resolution string

const (
	Merged      Resolution = "merged"
	LWW         Resolution = "lww"
	Quarantined Resolution = "quarantined"
)

type ConflictReport struct {
	Key        string
	Local      Versioned
	Remote     Versioned
	Resolution Resolution
	ResolvedAt time.Time
}

// --- Version vector algebra ----------------------------------------------

func CompareVectors(a, b VersionVector) Ordering {
	regions := make(map[RegionID]struct{}, len(a)+len(b))
	for r := range a { regions[r] = struct{}{} }
	for r := range b { regions[r] = struct{}{} }

	var aAhead, bAhead bool
	for r := range regions {
		if a[r] > b[r] { aAhead = true }
		if b[r] > a[r] { bAhead = true }
	}

	switch {
	case aAhead && bAhead:
		return Concurrent
	case aAhead:
		return After
	case bAhead:
		return Before
	default:
		return Equal
	}
}

func MergeVectors(a, b VersionVector) VersionVector {
	out := make(VersionVector, len(a)+len(b))
	for r, v := range a { out[r] = v }
	for r, v := range b {
		if v > out[r] { out[r] = v }
	}
	return out
}

func BumpVector(v VersionVector, region RegionID) VersionVector {
	out := make(VersionVector, len(v)+1)
	for r, n := range v { out[r] = n }
	out[region]++
	return out
}

// --- Region store ---------------------------------------------------------

type RegionStore struct {
	Region RegionID

	mu           sync.RWMutex
	data         map[string]Versioned
	log          []ReplicationRecord
	nextPosition uint64
	mergers      map[string]MergeFn
	quarantine   []ConflictReport
	appliedFrom  map[RegionID]uint64 // highest position applied from each peer
}

func NewRegionStore(region RegionID) *RegionStore {
	return &RegionStore{
		Region: region, data: make(map[string]Versioned), nextPosition: 1,
		mergers: make(map[string]MergeFn), appliedFrom: make(map[RegionID]uint64),
	}
}

func (s *RegionStore) RegisterMerger(keyPrefix string, fn MergeFn) {
	s.mu.Lock()
	defer s.mu.Unlock()
	s.mergers[keyPrefix] = fn
}

// Put is a local write: bump our own counter, append to the replication log.
func (s *RegionStore) Put(key string, value any) Versioned {
	s.mu.Lock()
	defer s.mu.Unlock()

	var base VersionVector
	if existing, ok := s.data[key]; ok { base = existing.Vector }

	entry := Versioned{
		Key: key, Value: value, Vector: BumpVector(base, s.Region),
		WrittenAt: time.Now().UTC(), OriginRegion: s.Region,
	}
	s.data[key] = entry
	s.log = append(s.log, ReplicationRecord{Entry: entry, LogPosition: s.nextPosition, EmittedAt: entry.WrittenAt})
	s.nextPosition++
	return entry
}

func (s *RegionStore) Get(key string) (Versioned, bool) {
	s.mu.RLock()
	defer s.mu.RUnlock()
	v, ok := s.data[key]
	return v, ok
}

// ChangesSince returns records a peer has not yet seen.
func (s *RegionStore) ChangesSince(position uint64, limit int) []ReplicationRecord {
	s.mu.RLock()
	defer s.mu.RUnlock()

	var out []ReplicationRecord
	for _, r := range s.log {
		if r.LogPosition <= position { continue }
		out = append(out, r)
		if limit > 0 && len(out) >= limit { break }
	}
	return out
}

// Apply handles inbound replication from a peer region.
func (s *RegionStore) Apply(record ReplicationRecord, from RegionID) *ConflictReport {
	s.mu.Lock()
	defer s.mu.Unlock()

	remote := record.Entry
	s.appliedFrom[from] = record.LogPosition

	local, exists := s.data[remote.Key]
	if !exists {
		s.data[remote.Key] = remote
		return nil
	}

	switch CompareVectors(local.Vector, remote.Vector) {
	case Before:
		s.data[remote.Key] = remote // remote strictly newer: take it
		return nil
	case After, Equal:
		return nil
	}

	// Genuinely concurrent — neither happened before the other.
	return s.resolveConflict(local, remote)
}

// resolveConflict must be called with the lock held.
func (s *RegionStore) resolveConflict(local, remote Versioned) *ConflictReport {
	merged := MergeVectors(local.Vector, remote.Vector)

	if fn := s.findMerger(local.Key); fn != nil {
		s.data[local.Key] = Versioned{
			Key: local.Key, Value: fn(local, remote), Vector: merged,
			WrittenAt: time.Now().UTC(), OriginRegion: s.Region,
		}
		report := newReport(local, remote, Merged)
		return &report
	}

	// No merger registered. Fall back to LWW with a deterministic tiebreak
	// so that every region converges on the SAME winner.
	remoteWins := remote.WrittenAt.After(local.WrittenAt) ||
		(remote.WrittenAt.Equal(local.WrittenAt) && remote.OriginRegion > local.OriginRegion)

	winner := local
	if remoteWins { winner = remote }
	winner.Vector = merged
	s.data[local.Key] = winner

	report := newReport(local, remote, LWW)
	// LWW discarded a real write: keep it for audit, never lose it silently.
	q := report
	q.Resolution = Quarantined
	s.quarantine = append(s.quarantine, q)
	return &report
}

func (s *RegionStore) findMerger(key string) MergeFn {
	for prefix, fn := range s.mergers {
		if strings.HasPrefix(key, prefix) { return fn }
	}
	return nil
}

func newReport(local, remote Versioned, res Resolution) ConflictReport {
	return ConflictReport{
		Key: local.Key, Local: local, Remote: remote,
		Resolution: res, ResolvedAt: time.Now().UTC(),
	}
}

func (s *RegionStore) PendingConflicts() []ConflictReport {
	s.mu.RLock()
	defer s.mu.RUnlock()
	out := make([]ConflictReport, len(s.quarantine))
	copy(out, s.quarantine)
	return out
}

// --- Replication pump with lag / RPO measurement --------------------------

type LagSample struct {
	FromRegion         RegionID
	ToRegion           RegionID
	PendingRecords     int
	OldestPendingAgeMs int64
	EstimatedRPOMs     int64
}

type Replicator struct {
	mu      sync.Mutex
	stores  map[RegionID]*RegionStore
	cursors map[string]uint64
}

func NewReplicator(stores map[RegionID]*RegionStore) *Replicator {
	return &Replicator{stores: stores, cursors: make(map[string]uint64)}
}

func cursorKey(from, to RegionID) string { return fmt.Sprintf("%s->%s", from, to) }

// Pump ships one batch from `from` to `to` and returns observed conflicts.
func (r *Replicator) Pump(from, to RegionID, batchSize int) ([]ConflictReport, error) {
	r.mu.Lock()
	source, okS := r.stores[from]
	target, okT := r.stores[to]
	key := cursorKey(from, to)
	cursor := r.cursors[key]
	r.mu.Unlock()

	if !okS || !okT { return nil, fmt.Errorf("unknown region pair %s->%s", from, to) }

	var conflicts []ConflictReport
	for _, record := range source.ChangesSince(cursor, batchSize) {
		if c := target.Apply(record, from); c != nil { conflicts = append(conflicts, *c) }
		r.mu.Lock()
		r.cursors[key] = record.LogPosition
		r.mu.Unlock()
	}
	return conflicts, nil
}

// MeasureLag treats replication lag as a position gap first, a time gap second.
func (r *Replicator) MeasureLag(from, to RegionID) (LagSample, error) {
	r.mu.Lock()
	source, ok := r.stores[from]
	cursor := r.cursors[cursorKey(from, to)]
	r.mu.Unlock()

	if !ok { return LagSample{}, fmt.Errorf("unknown region %s", from) }

	pending := source.ChangesSince(cursor, math.MaxInt32)
	sample := LagSample{FromRegion: from, ToRegion: to, PendingRecords: len(pending)}
	if len(pending) > 0 {
		// If `from` were lost right now, this is what `to` would not have.
		age := time.Since(pending[0].EmittedAt).Milliseconds()
		sample.OldestPendingAgeMs, sample.EstimatedRPOMs = age, age
	}
	return sample, nil
}