Multiplayer Game Matchmaking System.

# Matchmaker Leader Election with Fencing Tokens

import etcd3
from dataclasses import dataclass
from typing import Optional
import asyncio
import time


@dataclass
class LeaderLease:
    """Leadership lease with fencing token."""
    token: int
    expires_at: float
    
    def is_valid(self) -> bool:
        return time.time() < self.expires_at


class MatchmakerLeaderElection:
    """
    Leader election for matchmaker service.
    
    Ensures only ONE matchmaker runs per region.
    Uses fencing tokens to prevent zombie leaders.
    
    Week 5, Day 5: Leader Election concepts applied
    """
    
    def __init__(
        self,
        etcd_host: str,
        region: str,
        node_id: str,
        lease_ttl: int = 10
    ):
        self.client = etcd3.client(host=etcd_host)
        self.region = region
        self.node_id = node_id
        self.lease_ttl = lease_ttl
        self.election_key = f"/matchmaker/{region}/leader"
        
        self.lease: Optional[LeaderLease] = None
        self.etcd_lease = None
        self.is_leader = False
    
    async def campaign(self) -> Optional[int]:
        """
        Campaign for leadership.
        
        Returns fencing token if elected, None otherwise.
        """
        # Create lease
        self.etcd_lease = self.client.lease(self.lease_ttl)
        
        # Try to become leader (atomic compare-and-set)
        success, responses = self.client.transaction(
            compare=[
                self.client.transactions.version(self.election_key) == 0
            ],
            success=[
                self.client.transactions.put(
                    self.election_key,
                    self.node_id.encode(),
                    lease=self.etcd_lease
                )
            ],
            failure=[]
        )
        
        if success:
            # We're the leader - get fencing token from key version
            value, metadata = self.client.get(self.election_key)
            fencing_token = metadata.mod_revision  # Monotonically increasing
            
            self.lease = LeaderLease(
                token=fencing_token,
                expires_at=time.time() + self.lease_ttl
            )
            self.is_leader = True
            
            # Start keepalive
            asyncio.create_task(self._keepalive_loop())
            
            return fencing_token
        
        return None
    
    async def _keepalive_loop(self):
        """Keep lease alive while we're leader."""
        while self.is_leader:
            try:
                self.etcd_lease.refresh()
                self.lease.expires_at = time.time() + self.lease_ttl
                await asyncio.sleep(self.lease_ttl / 3)
            except Exception as e:
                self.is_leader = False
                break
    
    def get_fencing_token(self) -> Optional[int]:
        """Get current fencing token."""
        if self.lease and self.lease.is_valid():
            return self.lease.token
        return None
    
    async def resign(self):
        """Resign from leadership."""
        self.is_leader = False
        if self.etcd_lease:
            self.etcd_lease.revoke()


class FencedQueueOperations:
    """
    Queue operations protected by fencing tokens.
    
    Redis operations include fencing token validation.
    Prevents zombie leaders from corrupting state.
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
        # Track highest token seen per region. The {hash tag} keeps this key and
        # the queue key in the same Redis Cluster slot, as a multi-key script requires.
        self.highest_token_key = "matchmaker:highest_token:{{{region}}}"
    
    async def remove_players_from_queue(
        self,
        region: str,
        player_ids: list[str],
        fencing_token: int
    ) -> bool:
        """
        Remove players from queue atomically with fencing.
        
        Only succeeds if fencing_token >= highest seen.
        """
        # Lua script for atomic fencing check + removal
        script = """
        local highest_key = KEYS[1]
        local queue_key = KEYS[2]
        local token = tonumber(ARGV[1])
        local player_count = tonumber(ARGV[2])
        
        -- Check fencing token
        local highest = tonumber(redis.call('GET', highest_key) or '0')
        if token < highest then
            return 0  -- Stale token, reject
        end
        
        -- Update highest token
        redis.call('SET', highest_key, token)
        
        -- Remove players from queue
        for i = 3, 2 + player_count do
            redis.call('ZREM', queue_key, ARGV[i])
        end
        
        return 1  -- Success
        """
        
        # redis-py signature: eval(script, numkeys, *keys_and_args)
        result = await self.redis.eval(
            script,
            2,  # numkeys: the first two arguments after this are KEYS
            self.highest_token_key.format(region=region),
            f"queue:{{{region}}}",
            fencing_token, len(player_ids), *player_ids
        )
        
        return result == 1

# Rating Service with Strong Consistency

from dataclasses import dataclass
from typing import Optional, Dict, Tuple
from datetime import datetime, timedelta, timezone
import asyncpg
import redis.asyncio as redis  # async client: its commands are awaited


@dataclass
class RatingUpdate:
    """Result of a rating update."""
    player_id: str
    old_rating: int
    new_rating: int
    match_id: str
    updated_at: datetime


class RatingService:
    """
    Rating service with consistency guarantees.
    
    Week 5, Day 1: Consistency Models applied
    - Linearizable writes (always to primary)
    - Read-your-writes for recent updates
    - Causal consistency via version tracking
    """
    
    def __init__(
        self,
        primary_db: asyncpg.Pool,
        replica_db: asyncpg.Pool,
        redis_client: redis.Redis
    ):
        self.primary = primary_db
        self.replica = replica_db
        self.redis = redis_client
        
        # Track recent updates for read-your-writes
        self.recent_update_ttl = timedelta(seconds=30)
    
    async def update_ratings_for_match(
        self,
        match_id: str,
        winner_ids: list[str],
        loser_ids: list[str],
        is_draw: bool = False
    ) -> list[RatingUpdate]:
        """
        Update ratings for all players in a match.
        
        Uses transaction for atomicity.
        Linearizable consistency (writes to primary).
        """
        updates = []
        
        async with self.primary.acquire() as conn:
            async with conn.transaction():
                # Get current ratings
                all_player_ids = winner_ids + loser_ids
                ratings = await conn.fetch(
                    """
                    SELECT player_id, elo_rating 
                    FROM players 
                    WHERE player_id = ANY($1)
                    ORDER BY player_id  -- same lock order in every transaction avoids deadlocks
                    FOR UPDATE  -- Lock rows for consistent read-modify-write
                    """,
                    all_player_ids
                )
                
                rating_map = {r['player_id']: r['elo_rating'] for r in ratings}
                
                # Calculate new ratings (ELO formula)
                new_ratings = self._calculate_elo_changes(
                    winners={pid: rating_map[pid] for pid in winner_ids},
                    losers={pid: rating_map[pid] for pid in loser_ids},
                    is_draw=is_draw
                )
                
                # Update all ratings atomically
                for player_id, new_rating in new_ratings.items():
                    old_rating = rating_map[player_id]
                    
                    await conn.execute(
                        """
                        UPDATE players 
                        SET elo_rating = $1, 
                            last_match_at = NOW(),
                            rating_version = rating_version + 1
                        WHERE player_id = $2
                        """,
                        new_rating, player_id
                    )
                    
                    # Record update for read-your-writes
                    await self._record_recent_update(player_id)
                    
                    updates.append(RatingUpdate(
                        player_id=player_id,
                        old_rating=old_rating,
                        new_rating=new_rating,
                        match_id=match_id,
                        updated_at=datetime.now(timezone.utc)
                    ))
        
        return updates
    
    async def get_rating(
        self,
        player_id: str,
        require_fresh: bool = False
    ) -> Tuple[int, int]:
        """
        Get player's rating.
        
        Returns (rating, version).
        
        If require_fresh=True or player has recent update,
        reads from primary (linearizable).
        Otherwise reads from replica (eventually consistent).
        """
        # Check for recent update (read-your-writes)
        has_recent_update = await self._has_recent_update(player_id)
        
        if require_fresh or has_recent_update:
            # Read from primary for consistency
            result = await self.primary.fetchrow(
                "SELECT elo_rating, rating_version FROM players WHERE player_id = $1",
                player_id
            )
        else:
            # Read from replica (faster, eventual consistency)
            result = await self.replica.fetchrow(
                "SELECT elo_rating, rating_version FROM players WHERE player_id = $1",
                player_id
            )
        
        if result:
            return result['elo_rating'], result['rating_version']
        
        raise PlayerNotFoundError(player_id)
    
    async def get_ratings_for_matchmaking(
        self,
        player_ids: list[str]
    ) -> Dict[str, int]:
        """
        Get ratings for matchmaking.
        
        ALWAYS reads from primary for linearizable consistency.
        Matchmaker needs accurate ratings for fair matches.
        """
        results = await self.primary.fetch(
            "SELECT player_id, elo_rating FROM players WHERE player_id = ANY($1)",
            player_ids
        )
        
        return {r['player_id']: r['elo_rating'] for r in results}
    
    async def _record_recent_update(self, player_id: str):
        """Record that player has a recent rating update."""
        key = f"rating:recent_update:{player_id}"
        await self.redis.setex(
            key,
            int(self.recent_update_ttl.total_seconds()),
            "1"
        )
    
    async def _has_recent_update(self, player_id: str) -> bool:
        """Check if player has a recent rating update."""
        key = f"rating:recent_update:{player_id}"
        return bool(await self.redis.exists(key))
    
    def _calculate_elo_changes(
        self,
        winners: Dict[str, int],
        losers: Dict[str, int],
        is_draw: bool,
        k_factor: int = 32
    ) -> Dict[str, int]:
        """Calculate new ELO ratings using standard formula."""
        # Average ratings
        winner_avg = sum(winners.values()) / len(winners)
        loser_avg = sum(losers.values()) / len(losers)
        
        # Expected scores
        exp_winner = 1 / (1 + 10 ** ((loser_avg - winner_avg) / 400))
        exp_loser = 1 - exp_winner
        
        # Actual scores
        if is_draw:
            actual_winner = 0.5
            actual_loser = 0.5
        else:
            actual_winner = 1.0
            actual_loser = 0.0
        
        # Calculate changes
        new_ratings = {}
        
        for pid, rating in winners.items():
            change = k_factor * (actual_winner - exp_winner)
            new_ratings[pid] = max(0, min(3000, round(rating + change)))
        
        for pid, rating in losers.items():
            change = k_factor * (actual_loser - exp_loser)
            new_ratings[pid] = max(0, min(3000, round(rating + change)))
        
        return new_ratings

# Match Creation Saga

from dataclasses import dataclass, field
from typing import Optional, List, Dict
from enum import Enum
from datetime import datetime, timedelta, timezone
import asyncio


class MatchStatus(Enum):
    PENDING_CONFIRMATIONS = "pending_confirmations"
    ALLOCATING_SERVER = "allocating_server"
    CONFIRMED = "confirmed"
    CANCELLED = "cancelled"
    FAILED = "failed"


@dataclass
class MatchCandidate:
    """A potential match being formed."""
    match_id: str
    region: str
    game_mode: str
    team1_player_ids: List[str]
    team2_player_ids: List[str]
    status: MatchStatus
    created_at: datetime
    fencing_token: int
    
    # Saga state
    removed_from_queue: bool = False
    ready_checks_sent: bool = False
    confirmations: Dict[str, bool] = field(default_factory=dict)
    server_allocated: Optional[str] = None
    match_written: bool = False


class MatchCreationSaga:
    """
    Saga for creating a match.
    
    Week 5, Day 2: Saga Pattern with compensation
    
    Handles partial failures and ensures players
    are never left in limbo.
    """
    
    def __init__(
        self,
        queue_service,
        gateway_service,
        server_allocator,
        match_repository,
        fencing_token: int
    ):
        self.queue = queue_service
        self.gateway = gateway_service
        self.servers = server_allocator
        self.matches = match_repository
        self.fencing_token = fencing_token
    
    async def execute(self, candidate: MatchCandidate) -> MatchCandidate:
        """
        Execute the match creation saga.
        
        Returns the final state of the match candidate.
        """
        all_players = candidate.team1_player_ids + candidate.team2_player_ids
        
        try:
            # Step 1: Remove players from queue
            await self._step_remove_from_queue(candidate, all_players)
            
            # Step 2: Send ready checks
            await self._step_send_ready_checks(candidate, all_players)
            
            # Step 3: Wait for confirmations
            confirmed = await self._step_collect_confirmations(
                candidate, 
                all_players,
                timeout=timedelta(seconds=30)
            )
            
            if not confirmed:
                # Partial compensation - some players declined
                await self._compensate_unconfirmed(candidate)
                candidate.status = MatchStatus.CANCELLED
                return candidate
            
            # Step 4: Allocate game server (PIVOT POINT)
            try:
                server = await self._step_allocate_server(candidate)
            except NoServersAvailableError:
                # Full compensation - re-queue everyone
                await self._compensate_full(candidate, all_players)
                candidate.status = MatchStatus.FAILED
                return candidate
            
            # PIVOT PASSED - Forward recovery only from here
            candidate.server_allocated = server.address
            
            # Step 5: Write match to database (retry until success)
            await self._step_write_match(candidate)
            
            # Step 6: Notify players (retry until success)
            await self._step_notify_players(candidate, all_players)
            
            candidate.status = MatchStatus.CONFIRMED
            return candidate
            
        except Exception as e:
            if candidate.server_allocated:
                # Past the pivot: never compensate; a recovery job retries forward
                raise
            # Unexpected error before the pivot - full compensation
            await self._compensate_full(candidate, all_players)
            candidate.status = MatchStatus.FAILED
            raise
    
    async def _step_remove_from_queue(
        self,
        candidate: MatchCandidate,
        player_ids: List[str]
    ):
        """Step 1: Remove players from queue."""
        success = await self.queue.remove_players(
            region=candidate.region,
            player_ids=player_ids,
            fencing_token=self.fencing_token,
            pending_match_id=candidate.match_id
        )
        
        if not success:
            raise FencingError("Stale fencing token")
        
        candidate.removed_from_queue = True
    
    async def _step_send_ready_checks(
        self,
        candidate: MatchCandidate,
        player_ids: List[str]
    ):
        """Step 2: Send ready checks to all players."""
        await self.gateway.send_ready_checks(
            player_ids=player_ids,
            match_id=candidate.match_id,
            game_mode=candidate.game_mode
        )
        candidate.ready_checks_sent = True
    
    async def _step_collect_confirmations(
        self,
        candidate: MatchCandidate,
        player_ids: List[str],
        timeout: timedelta
    ) -> bool:
        """Step 3: Collect player confirmations."""
        deadline = datetime.now(timezone.utc) + timeout
        
        while datetime.now(timezone.utc) < deadline:
            # Get current confirmations
            confirmations = await self.gateway.get_confirmations(
                candidate.match_id
            )
            candidate.confirmations = confirmations
            
            # Check if all confirmed
            confirmed_count = sum(1 for c in confirmations.values() if c)
            declined_count = sum(1 for c in confirmations.values() if c is False)
            
            if confirmed_count == len(player_ids):
                return True
            
            if declined_count > 0:
                # Someone explicitly declined
                return False
            
            await asyncio.sleep(0.5)
        
        # Timeout - treat as decline
        return False
    
    async def _step_allocate_server(
        self,
        candidate: MatchCandidate
    ):
        """Step 4: Allocate game server (PIVOT)."""
        server = await self.servers.allocate(
            region=candidate.region,
            game_mode=candidate.game_mode,
            match_id=candidate.match_id
        )
        return server
    
    async def _step_write_match(self, candidate: MatchCandidate):
        """Step 5: Write match to database (forward recovery)."""
        max_retries = 10
        
        for attempt in range(max_retries):
            try:
                await self.matches.create(
                    match_id=candidate.match_id,
                    region=candidate.region,
                    game_mode=candidate.game_mode,
                    team1=candidate.team1_player_ids,
                    team2=candidate.team2_player_ids,
                    server_address=candidate.server_allocated
                )
                candidate.match_written = True
                return
            except Exception as e:
                if attempt == max_retries - 1:
                    raise
                await asyncio.sleep(2 ** attempt)  # Exponential backoff
    
    async def _step_notify_players(
        self,
        candidate: MatchCandidate,
        player_ids: List[str]
    ):
        """Step 6: Notify players of match (forward recovery)."""
        # Retry until all notified
        for _ in range(10):
            try:
                await self.gateway.notify_match_ready(
                    player_ids=player_ids,
                    match_id=candidate.match_id,
                    server_address=candidate.server_allocated
                )
                return
            except Exception:
                await asyncio.sleep(1)
    
    async def _compensate_unconfirmed(self, candidate: MatchCandidate):
        """Compensate for unconfirmed players."""
        # Re-queue players who confirmed
        confirmed_players = [
            pid for pid, confirmed in candidate.confirmations.items()
            if confirmed
        ]
        
        if confirmed_players:
            await self.queue.add_players_priority(
                region=candidate.region,
                player_ids=confirmed_players,
                priority=True  # Put at front of queue
            )
        
        # Notify confirmed players match was cancelled
        await self.gateway.notify_match_cancelled(
            player_ids=confirmed_players,
            reason="Other players did not confirm"
        )
    
    async def _compensate_full(
        self,
        candidate: MatchCandidate,
        player_ids: List[str]
    ):
        """Full compensation - re-queue all players."""
        if candidate.removed_from_queue:
            await self.queue.add_players_priority(
                region=candidate.region,
                player_ids=player_ids,
                priority=True
            )
        
        if candidate.ready_checks_sent:
            await self.gateway.notify_match_cancelled(
                player_ids=player_ids,
                reason="Match creation failed"
            )

# Party Queue with CRDT

from dataclasses import dataclass, field
from typing import Dict, Set, Optional
import json
import uuid
import time


@dataclass
class PartyMember:
    """Member in a party with metadata."""
    player_id: str
    joined_at: float
    is_leader: bool


class PartyORSet:
    """
    Party membership using OR-Set CRDT.
    
    Week 5, Day 4: Conflict Resolution with CRDTs
    
    Handles concurrent joins/leaves/kicks without conflicts.
    Add-wins semantics ensures players aren't accidentally removed.
    """
    
    def __init__(self, party_id: str, creator_id: str):
        self.party_id = party_id
        self.leader_id = creator_id
        
        # OR-Set: player_id -> {tag -> (tombstoned, joined_at, is_leader)}
        self.members: Dict[str, Dict[str, tuple]] = {}
        
        # Add creator as first member
        self._add_member(creator_id, is_leader=True)
    
    def _generate_tag(self) -> str:
        """Generate unique tag for this operation."""
        return f"{uuid.uuid4()}:{time.time()}"
    
    def _add_member(self, player_id: str, is_leader: bool = False):
        """Add a member with new unique tag."""
        tag = self._generate_tag()
        
        if player_id not in self.members:
            self.members[player_id] = {}
        
        self.members[player_id][tag] = (False, time.time(), is_leader)
    
    def add_member(self, player_id: str) -> bool:
        """Add a player to the party."""
        if self.get_member_count() >= 5:
            return False  # Party full (local check only: concurrent joins on
                          # different servers can exceed 5 after a merge, since
                          # CRDTs cannot enforce global invariants)
        
        self._add_member(player_id)
        return True
    
    def remove_member(self, player_id: str):
        """Remove a player (tombstone all their tags)."""
        if player_id in self.members:
            for tag in self.members[player_id]:
                tombstoned, joined_at, is_leader = self.members[player_id][tag]
                self.members[player_id][tag] = (True, joined_at, is_leader)
    
    def is_member(self, player_id: str) -> bool:
        """Check if player is currently a member."""
        if player_id not in self.members:
            return False
        
        # Member if any non-tombstoned tag exists
        return any(
            not tombstoned
            for tombstoned, _, _ in self.members[player_id].values()
        )
    
    def get_members(self) -> list[str]:
        """Get all current members."""
        return [
            pid for pid in self.members
            if self.is_member(pid)
        ]
    
    def get_member_count(self) -> int:
        """Get count of current members."""
        return len(self.get_members())
    
    def merge(self, other: 'PartyORSet'):
        """
        Merge with another party state.
        
        OR-Set merge:
        - Union of all tags
        - Tombstoned if tombstoned in either
        - Add wins: new tag from concurrent add survives
        """
        all_players = set(self.members.keys()) | set(other.members.keys())
        
        for player_id in all_players:
            self_tags = self.members.get(player_id, {})
            other_tags = other.members.get(player_id, {})
            
            merged_tags = {}
            all_tag_ids = set(self_tags.keys()) | set(other_tags.keys())
            
            for tag in all_tag_ids:
                if tag in self_tags and tag in other_tags:
                    # Both have this tag - tombstoned if either tombstoned
                    self_tomb, self_time, self_leader = self_tags[tag]
                    other_tomb, other_time, other_leader = other_tags[tag]
                    merged_tags[tag] = (
                        self_tomb or other_tomb,
                        max(self_time, other_time),
                        self_leader or other_leader
                    )
                elif tag in self_tags:
                    merged_tags[tag] = self_tags[tag]
                else:
                    merged_tags[tag] = other_tags[tag]
            
            if merged_tags:
                self.members[player_id] = merged_tags
    
    def to_dict(self) -> dict:
        """Serialize for storage/transmission."""
        return {
            "party_id": self.party_id,
            "leader_id": self.leader_id,
            "members": {
                pid: {
                    tag: {"tombstoned": t, "joined_at": j, "is_leader": l}
                    for tag, (t, j, l) in tags.items()
                }
                for pid, tags in self.members.items()
            }
        }
    
    @classmethod
    def from_dict(cls, data: dict) -> 'PartyORSet':
        """Deserialize from storage."""
        party = cls.__new__(cls)
        party.party_id = data["party_id"]
        party.leader_id = data["leader_id"]
        party.members = {
            pid: {
                tag: (info["tombstoned"], info["joined_at"], info["is_leader"])
                for tag, info in tags.items()
            }
            for pid, tags in data["members"].items()
        }
        return party


class PartyService:
    """
    Party management service with CRDT-based conflict resolution.
    """
    
    def __init__(self, redis_client):
        self.redis = redis_client
    
    async def create_party(self, creator_id: str) -> str:
        """Create a new party."""
        party_id = str(uuid.uuid4())
        party = PartyORSet(party_id, creator_id)
        
        await self._save_party(party)
        await self._set_player_party(creator_id, party_id)
        
        return party_id
    
    async def join_party(self, party_id: str, player_id: str) -> bool:
        """Join an existing party."""
        party = await self._load_party(party_id)
        if not party:
            return False
        
        if party.add_member(player_id):
            await self._save_party(party)
            await self._set_player_party(player_id, party_id)
            return True
        
        return False
    
    async def leave_party(self, party_id: str, player_id: str):
        """Leave a party."""
        party = await self._load_party(party_id)
        if not party:
            return
        
        party.remove_member(player_id)
        await self._save_party(party)
        await self._clear_player_party(player_id)
        
        # Check if party is now empty
        if party.get_member_count() == 0:
            await self._delete_party(party_id)
    
    async def merge_party_updates(
        self,
        party_id: str,
        remote_state: dict
    ):
        """
        Merge remote party state (from another server).
        
        Used for cross-region party sync.
        """
        local_party = await self._load_party(party_id)
        remote_party = PartyORSet.from_dict(remote_state)
        
        if local_party:
            local_party.merge(remote_party)
            await self._save_party(local_party)
        else:
            await self._save_party(remote_party)
    
    async def get_party_for_matchmaking(
        self,
        party_id: str
    ) -> Optional[list[str]]:
        """Get party members for matchmaking."""
        party = await self._load_party(party_id)
        if not party:
            return None
        
        return party.get_members()
    
    async def _save_party(self, party: PartyORSet):
        """Save party to Redis."""
        await self.redis.set(
            f"party:{party.party_id}",
            json.dumps(party.to_dict())
        )
    
    async def _load_party(self, party_id: str) -> Optional[PartyORSet]:
        """Load party from Redis."""
        data = await self.redis.get(f"party:{party_id}")
        if data:
            return PartyORSet.from_dict(json.loads(data))
        return None
    
    async def _delete_party(self, party_id: str):
        """Delete party from Redis."""
        await self.redis.delete(f"party:{party_id}")
    
    async def _set_player_party(self, player_id: str, party_id: str):
        """Set player's current party."""
        await self.redis.set(f"player_party:{player_id}", party_id)
    
    async def _clear_player_party(self, player_id: str):
        """Clear player's current party."""
        await self.redis.delete(f"player_party:{player_id}")