""" Zone, Dungeon Instance, and Secret Chamber Engine for FreeExile. Implements: 1. Authoritative Map & Zone Registry (Sanctuary, Open World, Dungeons, Secret Chambers). 2. Boundary-Checked Movement, Spawning, and Anti-Cheat Teleport Validation. 3. Waypoint Network, Proximity Discovery, and Fast Travel Gating. 4. Portal Traversal with Level and Quest Prerequisite Verification. 5. Dynamic Dungeon Instances with Layer Progression, Spatial Grid Isolation, and Time Limits. 6. Cryptic Secret Chamber Spawning, Lifespan Timers, and Auto-Collapse Mechanics. 7. Seamless Closed-Loop Integration with QuestEngine & Combat Subsystems. """ from __future__ import annotations import uuid from typing import Dict, List, Optional, Set, Tuple from server.world.spatial_grid import SpatialGrid from server.common.math_utils import euclidean_distance_2d, is_within_radius_2d from server.common.time_utils import now_ms from server.world.zone_types import ( ZoneType, ZoneEnvironment, Waypoint, ZonePortal, ZoneDefinition, PlayerLocation, DungeonTemplate, InstanceSession, SecretChamberTemplate, SecretChamberSession, ) from server.world.zone_catalog import register_canonical_zones_and_templates __all__ = [ "ZoneEngine", "ZoneType", "ZoneEnvironment", "Waypoint", "ZonePortal", "ZoneDefinition", "PlayerLocation", "DungeonTemplate", "InstanceSession", "SecretChamberTemplate", "SecretChamberSession", ] class ZoneEngine: """ Authoritative Zone, Instance, and Secret Chamber Management Engine. Guarantees strict spatial bounds, zero-trust fast travel, isolated instances, and reactive secret chamber mechanics. """ def __init__(self) -> None: self.zones: Dict[str, ZoneDefinition] = {} self.dungeon_templates: Dict[str, DungeonTemplate] = {} self.secret_chamber_templates: Dict[str, SecretChamberTemplate] = {} # Player states self.player_locations: Dict[str, PlayerLocation] = {} self.player_unlocked_waypoints: Dict[str, Set[str]] = {} # Active dynamic sessions self.active_instances: Dict[str, InstanceSession] = {} self.active_secret_chambers: Dict[str, SecretChamberSession] = {} # Spatial grids for open world / sanctuary zones self.zone_spatial_grids: Dict[str, SpatialGrid] = {} # Register canonical maps, dungeons, and chambers from catalog register_canonical_zones_and_templates(self) def add_zone(self, zone: ZoneDefinition) -> None: """Registers a zone and initializes its spatial grid.""" self.zones[zone.zone_id] = zone self.zone_spatial_grids[zone.zone_id] = SpatialGrid(cell_size=64.0) # ------------------------------------------------------------------------- # LOOKUP AND INSPECTION APIS # ------------------------------------------------------------------------- def get_zone(self, zone_id: str) -> ZoneDefinition: if zone_id not in self.zones: raise KeyError(f"Zone '{zone_id}' not found in registry.") return self.zones[zone_id] def get_zone_definition(self, zone_id: str) -> Optional[ZoneDefinition]: """Safe lookup alias for zone definition; returns None if not found.""" return self.zones.get(zone_id) def get_dungeon_template(self, template_id: str) -> DungeonTemplate: if template_id not in self.dungeon_templates: raise KeyError(f"Dungeon template '{template_id}' not found.") return self.dungeon_templates[template_id] def get_secret_chamber_template(self, template_id: str) -> SecretChamberTemplate: if template_id not in self.secret_chamber_templates: raise KeyError(f"Secret chamber template '{template_id}' not found.") return self.secret_chamber_templates[template_id] def can_spawn_hostile_monsters(self, zone_id: str) -> bool: """ PoE2 Rule: Safe Havens (Sanctuary Hub and Player Hideout) prohibit hostile mobs. Only OPEN_WORLD, DUNGEON_INSTANCE, and SECRET_CHAMBER allow hostile encounters. """ zone = self.zones.get(zone_id) if zone is None: return False return zone.zone_type in (ZoneType.OPEN_WORLD, ZoneType.DUNGEON_INSTANCE, ZoneType.SECRET_CHAMBER) def is_in_waypoint_safe_radius(self, zone_id: str, x: float, y: float) -> bool: """ PoE2 Rule: Checks if coordinates (x, y) fall within the safe_radius of any registered waypoint in the specified zone. """ zone = self.zones.get(zone_id) if zone is None: return False for wp in zone.waypoints.values(): if is_within_radius_2d(x, y, wp.x, wp.y, wp.safe_radius): return True return False def validate_monster_spawn( self, zone_id: str, is_dummy: bool = False, x: Optional[float] = None, y: Optional[float] = None, tx: Optional[int] = None, ty: Optional[int] = None, ) -> Tuple[bool, str]: """Validates monster spawn in zone: checks Safe Havens, waypoint safe_radius, and tile passability.""" if isinstance(is_dummy, (int, float)) and type(is_dummy) is not bool: tx, ty = int(is_dummy), int(x) if x is not None else 0 is_dummy, x, y = False, None, None if is_dummy: return True, "Khôi Lỗi Luyện Võ (Target Dummy) được phép kích hoạt để thử nghiệm chiêu thức." if not self.can_spawn_hostile_monsters(zone_id): return False, f"Nghiêm cấm sản sinh quái vật trong khu vực An Toàn (Sanctuary/Hideout: [{zone_id}]) theo chuẩn PoE2!" if x is not None and y is not None: zone = self.zones.get(zone_id) if zone is not None: for wp in zone.waypoints.values(): if is_within_radius_2d(x, y, wp.x, wp.y, wp.safe_radius): return ( False, f"Nghiêm cấm sản sinh quái vật trong phạm vi an toàn ({wp.safe_radius}m) của điểm dịch chuyển [{wp.name}]!", ) if tx is not None and ty is not None: from server.world.procedural_map_engine import ProceduralMapEngine from server.world.map_data_types import TileType tile_type = ProceduralMapEngine.get_tile_type(zone_id, tx, ty) if tile_type in (TileType.WALL, TileType.CHASM, TileType.WATER, TileType.BOSS_GATE, TileType.VOID) or not tile_type.is_passable(): return ( False, f"Nghiêm cấm sản sinh quái vật trên địa hình không thể đi qua [{tile_type.name}] tại ({tx}, {ty})!", ) return True, f"Khu vực [{zone_id}] hợp lệ để sản sinh quái vật dã ngoại." def get_player_location(self, player_id: str) -> PlayerLocation: if player_id not in self.player_locations: self.spawn_player(player_id, "zone_boundless_sanctuary") return self.player_locations[player_id] def get_unlocked_waypoints(self, player_id: str) -> Set[str]: if player_id not in self.player_unlocked_waypoints: default_unlocked: Set[str] = set() for z in self.zones.values(): for wp in z.waypoints.values(): if wp.is_unlocked_by_default: default_unlocked.add(wp.waypoint_id) self.player_unlocked_waypoints[player_id] = default_unlocked return self.player_unlocked_waypoints[player_id] # ------------------------------------------------------------------------- # PLAYER MOVEMENT & BOUNDS CHECKING # ------------------------------------------------------------------------- def spawn_player( self, player_id: str, zone_id: str, x: Optional[float] = None, y: Optional[float] = None ) -> PlayerLocation: zone = self.get_zone(zone_id) spawn_x = x if x is not None else zone.default_spawn_x spawn_y = y if y is not None else zone.default_spawn_y loc = PlayerLocation( player_id=player_id, zone_id=zone_id, x=spawn_x, y=spawn_y, z=0.0, instance_id=None, last_update_timestamp_ms=now_ms(), ) self.player_locations[player_id] = loc self.get_unlocked_waypoints(player_id) return loc def update_player_position( self, player_id: str, new_x: float, new_y: float ) -> Tuple[bool, str]: loc = self.get_player_location(player_id) zone = self.get_zone(loc.zone_id) half_w = zone.bounds_width / 2.0 half_h = zone.bounds_height / 2.0 if abs(new_x) > half_w or abs(new_y) > half_h: return False, f"Vượt quá ranh giới bản đồ {zone.name} (Tọa độ: x={new_x:.1f}, y={new_y:.1f})" loc.x = new_x loc.y = new_y loc.last_update_timestamp_ms = now_ms() return True, "" # ------------------------------------------------------------------------- # WAYPOINTS & FAST TRAVEL # ------------------------------------------------------------------------- def check_waypoint_proximity( self, player_id: str, current_x: float, current_y: float ) -> Set[str]: """Checks if the player is within range of any locked waypoint and unlocks it.""" loc = self.get_player_location(player_id) zone = self.get_zone(loc.zone_id) unlocked_set = self.get_unlocked_waypoints(player_id) newly_discovered: Set[str] = set() for wp_id, wp in zone.waypoints.items(): if is_within_radius_2d(current_x, current_y, wp.x, wp.y, wp.discovery_radius): if wp_id not in unlocked_set: unlocked_set.add(wp_id) newly_discovered.add(wp_id) return newly_discovered def teleport_via_waypoint(self, player_id: str, target_waypoint_id: str) -> Tuple[bool, str]: """Teleports player to an unlocked waypoint.""" unlocked = self.get_unlocked_waypoints(player_id) if target_waypoint_id not in unlocked: return False, f"Chưa kích hoạt Trụ Thần Hành '{target_waypoint_id}'!" target_wp: Optional[Waypoint] = None for z in self.zones.values(): if target_waypoint_id in z.waypoints: target_wp = z.waypoints[target_waypoint_id] break if not target_wp: return False, f"Không tìm thấy Trụ Thần Hành '{target_waypoint_id}'." loc = self.get_player_location(player_id) loc.zone_id = target_wp.zone_id loc.x = target_wp.x loc.y = target_wp.y loc.z = target_wp.z loc.instance_id = None loc.last_update_timestamp_ms = now_ms() return True, f"Dịch chuyển thành công đến {target_wp.name}." # ------------------------------------------------------------------------- # PORTAL TRAVERSAL & LEVEL GATES # ------------------------------------------------------------------------- def traverse_portal( self, player_id: str, portal_id: str, player_level: int = 1, completed_quests: Optional[Set[str]] = None ) -> Tuple[bool, str, str]: """ Traverses a zone boundary portal. Returns: (success, target_zone_id, status_message) """ loc = self.get_player_location(player_id) current_zone = self.get_zone(loc.zone_id) if portal_id not in current_zone.portals: return False, "", f"Không tìm thấy Lối Qua Vùng Đất Chết '{portal_id}' trong bản đồ hiện tại." portal = current_zone.portals[portal_id] if player_level < portal.min_level: return False, "", f"Chưa đạt cấp độ yêu cầu ({player_level}/{portal.min_level}) để vào khu vực này!" if portal.required_quest_id and (completed_quests is None or portal.required_quest_id not in completed_quests): return False, "", f"Cần hoàn thành nhiệm vụ tiền đề '{portal.required_quest_id}' để mở phong ấn lối qua!" loc.zone_id = portal.target_zone_id loc.x = portal.target_x loc.y = portal.target_y loc.instance_id = None loc.last_update_timestamp_ms = now_ms() target_zone = self.get_zone(portal.target_zone_id) return True, portal.target_zone_id, f"Bước vào {target_zone.name} thành công." # ------------------------------------------------------------------------- # DUNGEON INSTANCE LIFECYCLE # ------------------------------------------------------------------------- def create_dungeon_instance( self, template_id: str, leader_id: str, members: List[str], leader_level: int, completed_quests: Set[str], ) -> Tuple[bool, Optional[InstanceSession], str]: """Creates an authoritative isolated dungeon instance.""" if template_id not in self.dungeon_templates: return False, None, f"Không tìm thấy mẫu phó bản '{template_id}'." tmpl = self.dungeon_templates[template_id] if leader_level < tmpl.min_level: return False, None, f"Chưa đạt cấp độ {tmpl.min_level} để mở {tmpl.name}." if tmpl.required_quest_id and tmpl.required_quest_id not in completed_quests: return False, None, f"Cần hoàn thành nhiệm vụ '{tmpl.required_quest_id}' trước khi tham chiến." all_members = [leader_id] + [m for m in members if m != leader_id] if len(all_members) > tmpl.max_party_size: return False, None, f"Số lượng người chơi ({len(all_members)}) vượt quá giới hạn ({tmpl.max_party_size})." instance_id = f"inst_{template_id}_{uuid.uuid4().hex[:8]}" instance_zone_id = f"zone_{template_id}" session = InstanceSession( instance_id=instance_id, template_id=template_id, instance_zone_id=instance_zone_id, leader_player_id=leader_id, member_player_ids=members, current_layer=1, max_layers=tmpl.max_layers, lives_remaining=tmpl.revive_lives, time_limit_sec=tmpl.time_limit_sec, start_timestamp_ms=now_ms(), spatial_grid=SpatialGrid(cell_size=64.0), ) self.active_instances[instance_id] = session for p in all_members: loc = self.get_player_location(p) loc.zone_id = instance_zone_id loc.instance_id = instance_id loc.x = 0.0 loc.y = 0.0 loc.last_update_timestamp_ms = now_ms() return True, session, f"Khai mở phó bản {tmpl.name} thành công!" def advance_instance_layer(self, instance_id: str) -> Tuple[int, bool]: """ Advances the instance to the next floor/layer. Returns: (new_layer, is_completed) """ if instance_id not in self.active_instances: raise KeyError(f"Instance '{instance_id}' not found.") session = self.active_instances[instance_id] if session.current_layer >= session.max_layers: session.is_completed = True return session.current_layer, True session.current_layer += 1 is_done = (session.current_layer >= session.max_layers) if is_done: session.is_completed = True return session.current_layer, is_done def close_instance(self, instance_id: str) -> None: """Evacuates all members back to sanctuary and destroys the instance session.""" if instance_id not in self.active_instances: return session = self.active_instances.pop(instance_id) current_time_ms = now_ms() for p in session.all_players: if p in self.player_locations: loc = self.player_locations[p] loc.zone_id = "zone_boundless_sanctuary" loc.instance_id = None loc.x = 0.0 loc.y = 0.0 loc.last_update_timestamp_ms = current_time_ms # ------------------------------------------------------------------------- # SECRET CHAMBER MECHANICS & AUTO-COLLAPSE # ------------------------------------------------------------------------- def spawn_secret_chamber( self, template_id: str, player_id: str, parent_zone_id: str, portal_x: float, portal_y: float, lifespan_sec: Optional[int] = None, ) -> Tuple[bool, Optional[SecretChamberSession], str]: """Dynamically spawns a secret chamber portal upon satisfying cryptic criteria.""" if template_id not in self.secret_chamber_templates: return False, None, f"Không tìm thấy mẫu mật thất '{template_id}'." tmpl = self.secret_chamber_templates[template_id] duration = lifespan_sec if lifespan_sec is not None else tmpl.lifespan_sec current_now_ms = now_ms() chamber_id = f"chamber_{template_id}_{uuid.uuid4().hex[:8]}" chamber_zone_id = f"zone_{template_id}" session = SecretChamberSession( chamber_id=chamber_id, template_id=template_id, chamber_zone_id=chamber_zone_id, player_id=player_id, portal_zone_id=parent_zone_id, portal_x=portal_x, portal_y=portal_y, spawn_timestamp_ms=current_now_ms, collapse_timestamp_ms=current_now_ms + (duration * 1000), is_active=True, is_discovered=True, is_cleared=False, spatial_grid=SpatialGrid(cell_size=64.0), ) self.active_secret_chambers[chamber_id] = session return True, session, f"Kỳ ngộ xuất hiện: Cổng vào {tmpl.name} đã khai mở!" def enter_secret_chamber(self, player_id: str, chamber_id: str) -> Tuple[bool, str]: """Player steps through the secret portal into the secret chamber.""" if chamber_id not in self.active_secret_chambers: return False, "Cổng mật thất không tồn tại hoặc đã sụp đổ!" chamber = self.active_secret_chambers[chamber_id] current_now_ms = now_ms() if current_now_ms >= chamber.collapse_timestamp_ms or not chamber.is_active: self.collapse_secret_chamber(chamber_id) return False, "Mật thất đã sụp đổ, lối vào đã khép lại!" if chamber.player_id != player_id: return False, "Cơ duyên mật thất này chỉ dành riêng cho người hữu duyên kích hoạt!" loc = self.get_player_location(player_id) loc.zone_id = chamber.chamber_zone_id loc.instance_id = chamber.chamber_id loc.x = 0.0 loc.y = 0.0 loc.last_update_timestamp_ms = current_now_ms tmpl = self.get_secret_chamber_template(chamber.template_id) return True, f"Bước vào {tmpl.name} thành công. Hãy nhanh chóng tìm kiếm cơ duyên!" def complete_secret_chamber(self, player_id: str, chamber_id: str) -> Tuple[bool, str]: """Marks the secret chamber objective as completed.""" if chamber_id not in self.active_secret_chambers: return False, "Mật thất không tồn tại!" chamber = self.active_secret_chambers[chamber_id] if chamber.player_id != player_id: return False, "Không phải chủ sở hữu mật thất!" chamber.is_cleared = True return True, "Khai mở trọn vẹn cơ duyên mật thất!" def collapse_secret_chamber(self, chamber_id: str) -> None: """Collapses the secret chamber and returns the player back to the parent zone entrance.""" if chamber_id not in self.active_secret_chambers: return chamber = self.active_secret_chambers.pop(chamber_id) chamber.is_active = False player_id = chamber.player_id if player_id in self.player_locations: loc = self.player_locations[player_id] loc.zone_id = chamber.portal_zone_id loc.instance_id = None loc.x = chamber.portal_x loc.y = chamber.portal_y loc.last_update_timestamp_ms = now_ms()