""" Authoritative Safe Teleportation Channel & Destination Validation Service for FreeExile. Implements the 3.5-second channeling state machine with instant movement and damage interrupt triggers, alongside the 5-Gate Destination Validation and spiral ray safe landing coordinate resolver. """ from __future__ import annotations import math import time from typing import Any, Callable, Dict, Optional, Tuple from server.world.town_portal_types import ( InterruptReason, TargetType, TeleportChannelResult, TeleportChannelSession, TeleportChannelStatus, ) class TeleportService: """Authoritative service managing safe player-to-player and portal teleportation.""" def __init__( self, party_service: Optional[Any] = None, social_service: Optional[Any] = None, guild_service: Optional[Any] = None, ) -> None: self.party_service = party_service self.social_service = social_service self.guild_service = guild_service self.active_channels: Dict[str, TeleportChannelSession] = {} self.zone_states: Dict[str, Dict[str, Any]] = {} self._channel_counter: int = 0 def set_zone_state( self, zone_id: str, is_boss_active: bool = False, is_secret_active: bool = False, player_count: int = 1, max_capacity: int = 6, is_active: bool = True, ) -> None: """Register or update runtime spatial capacity and quarantine states for a zone.""" self.zone_states[zone_id] = { "is_boss_active": is_boss_active, "is_secret_active": is_secret_active, "player_count": player_count, "max_capacity": max_capacity, "is_active": is_active, } def validate_destination_5_gates( self, player_id: str, target_id: str, target_type: TargetType, target_zone_id: str, target_alive: bool = True, ) -> Tuple[bool, str]: """Evaluate the 5 authoritative security gates before initiating or completing teleport.""" # Gate 1: Target Liveness if not target_alive: return False, "GATE_1_TARGET_DEAD: Mục tiêu đã tử vong hoặc không trực tuyến" # Gate 2: Relationship Authorization if target_type == TargetType.PARTY_MEMBER: if not self.party_service: return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Dịch vụ tổ đội không khả dụng" party_map = getattr(self.party_service, "player_party_map", {}) p_id = party_map.get(player_id) parties = getattr(self.party_service, "parties", {}) party = parties.get(p_id) if p_id else None if not party or target_id not in party.members: return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Không cùng tổ đội" elif target_type == TargetType.FRIEND: if not self.social_service or not self.social_service.is_friend(player_id, target_id): return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Không nằm trong danh sách bạn bè" elif target_type == TargetType.GUILDMATE: if not self.guild_service: return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Dịch vụ bang hội không khả dụng" guilds = getattr(self.guild_service, "guilds", {}) found = any(m.player_id == target_id for roster in guilds.values() for m in roster) if not found: return False, "GATE_2_RELATIONSHIP_UNAUTHORIZED: Không cùng bang hội" # Gate 3: Zone Capacity Limit z = self.zone_states.get( target_zone_id, {"player_count": 0, "max_capacity": 6, "is_active": True}, ) if z.get("player_count", 0) >= z.get("max_capacity", 6): return False, "GATE_3_CAPACITY_EXCEEDED: Khu vực đã đầy người chơi" # Gate 4: Zone Lifecycle if not z.get("is_active", True): return False, "GATE_4_LIFECYCLE_CLOSED: Khu vực đã đóng hoặc hết hạn" # Gate 5: Restricted Zone Isolation & Anti-Exploit Quarantine if z.get("is_boss_active", False): return False, "GATE_5_QUARANTINE_BOSS: Đang trong giao tranh Boss nguy cấp" if z.get("is_secret_active", False): return False, "GATE_5_QUARANTINE_SECRET: Mật thất thử thách đang diễn ra" # Strict check for private hideout isolation if target_zone_id.startswith("zone_player_hideout_") and not target_zone_id.endswith(player_id): return False, "GATE_5_RESTRICTED_ISOLATION: Sào huyệt riêng tư của người chơi khác" return True, "SUCCESS" def start_teleport_channel( self, player_id: str, target_type: TargetType, target_id: str, target_zone_id: str, target_alive: bool = True, initial_x: float = 0.0, initial_y: float = 0.0, ) -> TeleportChannelResult: """Validate destination and start authoritative 3.5s channeling session.""" ok, reason = self.validate_destination_5_gates( player_id=player_id, target_id=target_id, target_type=target_type, target_zone_id=target_zone_id, target_alive=target_alive, ) if not ok: return TeleportChannelResult(False, reason) self._channel_counter += 1 channel_id = f"chan_{player_id}_{int(time.time() * 1000)}_{self._channel_counter}" session = TeleportChannelSession( session_id=channel_id, player_id=player_id, target_type=target_type, target_id=target_id, start_time=time.time(), duration_seconds=3.5, initial_x=initial_x, initial_y=initial_y, status=TeleportChannelStatus.CHANNELING, target_zone_id=target_zone_id, ) self.active_channels[player_id] = session return TeleportChannelResult(True, channel_id) def on_player_moved(self, player_id: str, mag: float) -> Optional[InterruptReason]: """Trigger immediate channel cancellation if movement magnitude exceeds threshold (>0.05).""" if player_id in self.active_channels and mag > 0.05: del self.active_channels[player_id] return InterruptReason.INTERRUPT_MOVEMENT return None def on_player_damaged(self, player_id: str, damage: float) -> Optional[InterruptReason]: """Trigger immediate channel cancellation if player receives incoming damage (>0.0).""" if player_id in self.active_channels and damage > 0.0: del self.active_channels[player_id] return InterruptReason.INTERRUPT_DAMAGE return None def cancel_channel(self, player_id: str) -> bool: """Manually cancel an active channeling session (e.g. dodging or manual abort).""" if player_id in self.active_channels: del self.active_channels[player_id] return True return False def complete_channel(self, player_id: str, elapsed_time: float) -> Tuple[bool, str]: """Authoritatively complete channeling session and return verified target zone ID.""" chan = self.active_channels.get(player_id) if not chan: return False, "NO_ACTIVE_CHANNEL" if elapsed_time < chan.duration_seconds: return False, "CAST_INCOMPLETE" # Post-flight gate check ok, reason = self.validate_destination_5_gates( player_id=player_id, target_id=chan.target_id, target_type=chan.target_type, target_zone_id=chan.target_zone_id, target_alive=True, ) if not ok: del self.active_channels[player_id] return False, reason dest_zone = chan.target_zone_id del self.active_channels[player_id] return True, dest_zone def resolve_safe_landing_coords( self, target_x: float, target_y: float, blocked_checker: Optional[Callable[[float, float], bool]] = None, ) -> Tuple[float, float]: """Resolve safe landing coordinates using spiral ray search (2.0m - 3.0m) avoiding collisions.""" angles = [ 0.0, math.pi / 4, math.pi / 2, 3 * math.pi / 4, math.pi, 5 * math.pi / 4, 3 * math.pi / 2, 7 * math.pi / 4, ] for radius in (2.0, 2.5, 3.0): for angle in angles: cand_x = round(target_x + radius * math.cos(angle), 2) cand_y = round(target_y + radius * math.sin(angle), 2) if blocked_checker is None or not blocked_checker(cand_x, cand_y): return cand_x, cand_y return round(target_x + 2.0, 2), round(target_y, 2)