""" FreeExile Social Service: High-Level Business Logic for Friendships & Presence. Coordinates friendship lifecycles, mutual block verification, presence sync, and Chat ChannelManager blacklist integration. """ from __future__ import annotations import time from typing import Any, Dict, List, Optional, Set from server.social.social_repository import SocialRepository from server.social.social_types import ( FriendDTO, FriendshipStatus, PresenceStatus, SocialActionResponse, ) class SocialService: """Core Social Engine managing friend relations, real-time presence, and blocklists.""" def __init__( self, repository: Optional[SocialRepository] = None, channel_manager: Optional[Any] = None, ) -> None: self._repository = repository or SocialRepository() self._channel_manager = channel_manager self.friend_requests: Dict[str, Set[str]] = {} self._presence_cache: Dict[str, FriendDTO] = {} self._sync_friend_requests_from_db() def _sync_friend_requests_from_db(self) -> None: """Hydrates active pending friend requests from repository into memory cache.""" pass def is_blocked(self, sender_id: str, target_id: str) -> bool: """Returns True if either user has blocked the other (mutual block).""" return self._repository.is_blocked(sender_id, target_id) def is_friend(self, player_id: str, target_id: str) -> bool: """Returns True if players are mutual friends and not blocked.""" return self._repository.is_friend(player_id, target_id) def block_player(self, player_id: str, target_id: str) -> None: """Blocks target player, cleans friend state, and syncs chat blacklist.""" self._repository.block_player(player_id, target_id) self.friend_requests.get(player_id, set()).discard(target_id) self.friend_requests.get(target_id, set()).discard(player_id) if self._channel_manager is not None: try: pid = int(player_id) tid = int(target_id) self._channel_manager.set_blacklist_entry(pid, tid, is_blocked=True) except (ValueError, TypeError): if hasattr(self._channel_manager, "_player_blacklists"): self._channel_manager._player_blacklists.setdefault(player_id, set()).add(target_id) def unblock_player(self, player_id: str, target_id: str) -> None: """Unblocks target player and updates chat blacklist.""" self._repository.unblock_player(player_id, target_id) if self._channel_manager is not None: try: pid = int(player_id) tid = int(target_id) self._channel_manager.set_blacklist_entry(pid, tid, is_blocked=False) except (ValueError, TypeError): if hasattr(self._channel_manager, "_player_blacklists"): self._channel_manager._player_blacklists.get(player_id, set()).discard(target_id) def send_friend_request(self, sender_id: str, target_id: str) -> bool: """Sends an outgoing friend request to target_id.""" if sender_id == target_id: return False if self.is_blocked(sender_id, target_id) or self.is_friend(sender_id, target_id): return False ok = self._repository.add_friend_request(sender_id, target_id) if ok: self.friend_requests.setdefault(target_id, set()).add(sender_id) return ok def accept_friend_request(self, target_id: str, sender_id: str) -> bool: """Accepts a pending friend request sent by sender_id to target_id.""" if self.is_blocked(target_id, sender_id): return False ok = self._repository.accept_friend_request(target_id, sender_id) if ok: self.friend_requests.get(target_id, set()).discard(sender_id) return ok def decline_friend_request(self, target_id: str, sender_id: str) -> bool: """Declines an incoming friend request from sender_id.""" ok = self._repository.decline_friend_request(target_id, sender_id) if ok: self.friend_requests.get(target_id, set()).discard(sender_id) return ok def remove_friend(self, player_id: str, friend_id: str) -> bool: """Removes an existing friendship.""" return self._repository.remove_friend(player_id, friend_id) def set_friend_note(self, owner_id: str, friend_id: str, note: str) -> bool: """Sets a custom personal note for an accepted friend.""" return self._repository.set_friend_note(owner_id, friend_id, note) def update_presence( self, player_id: str, name: str, level: int, is_online: bool, zone_id: str, zone_name: str, ) -> None: """Updates live presence metadata in the in-memory cache.""" status = PresenceStatus.ONLINE if is_online else PresenceStatus.OFFLINE self._presence_cache[player_id] = FriendDTO( player_id=player_id, character_name=name, level=level, character_class="Outcast", presence_status=status, zone_id=zone_id, current_map_name=zone_name, friend_note="", last_seen_timestamp=time.time(), ) def update_player_presence( self, player_id: str, level: int, zone_id: str, presence_status: PresenceStatus, character_name: str = "", character_class: str = "Outcast", current_map_name: str = "", ) -> None: """Canonical presence update adhering to dispatch specification.""" existing = self._presence_cache.get(player_id) name = character_name or (existing.character_name if existing else player_id) cls = character_class or (existing.character_class if existing else "Outcast") map_name = current_map_name or (existing.current_map_name if existing else zone_id) self._presence_cache[player_id] = FriendDTO( player_id=player_id, character_name=name, level=level, character_class=cls, presence_status=presence_status, zone_id=zone_id, current_map_name=map_name, friend_note="", last_seen_timestamp=time.time(), ) def get_friend_presence(self, player_id: str) -> List[FriendDTO]: """Retrieves list of mutual friends with live presence info and notes.""" db_friends = self._repository.get_friends(player_id) result: List[FriendDTO] = [] for f in db_friends: fid = f["friend_id"] if self.is_blocked(player_id, fid): continue note = f.get("friend_note", "") cached = self._presence_cache.get(fid) if cached: dto = FriendDTO( player_id=fid, character_name=cached.character_name, level=cached.level, character_class=cached.character_class, presence_status=cached.presence_status, zone_id=cached.zone_id, current_map_name=cached.current_map_name, friend_note=note, last_seen_timestamp=cached.last_seen_timestamp, ) else: dto = FriendDTO( player_id=fid, character_name=fid, level=1, character_class="Outcast", presence_status=PresenceStatus.OFFLINE, zone_id="offline", current_map_name="Offline", friend_note=note, last_seen_timestamp=0.0, ) result.append(dto) return result def get_friends(self, player_id: str) -> List[FriendDTO]: """Convenience alias for get_friend_presence.""" return self.get_friend_presence(player_id) def get_blocked_list(self, player_id: str) -> List[str]: """Retrieves list of player IDs blocked by player_id.""" return self._repository.get_blocked_list(player_id)