""" Main Authoritative Chat Microservice for FreeExile. Coordinates Two-Tier Moderation, HMAC Item Linking, Channel Policy, and 1M CCU Fan-out Router. Completely isolated from the 30Hz World Simulation Engine to ensure Zero-Latency. """ from __future__ import annotations import time import uuid from typing import List, Optional, Tuple from server.chat.channel_manager import ChannelManager from server.chat.chat_cluster_router import ChatClusterRouter, SubscriberCallback from server.chat.chat_types import ( ChatChannelType, ChatMessageDTO, ItemSnapshotDTO, QueryItemSnapshotResponseDTO, SendChatRequestDTO, SendChatResponseDTO, ) from server.chat.item_link_service import ItemLinkService from server.chat.moderation import ChatModerationPipeline class ChatService: """ Facade and core orchestration engine for FreeExile's Chat System. Decoupled 100% from the 30Hz Zone Server simulation loop. """ def __init__( self, item_secret_key: bytes = b"FREEEXILE_SECRET_SIGNING_KEY_2026", cluster_shards: int = 64 ) -> None: self.moderator = ChatModerationPipeline() self.item_link_service = ItemLinkService(secret_signing_key=item_secret_key) self.channel_manager = ChannelManager() self.cluster_router = ChatClusterRouter(num_shards=cluster_shards) async def handle_send_chat(self, request: SendChatRequestDTO) -> SendChatResponseDTO: """ Processes an incoming chat message request end-to-end: 1. Channel permission, rate limit, and persistent mute verification. 2. Two-tier content moderation and Anti-RMT risk scoring. 3. Linked items verification (HMAC check). 4. Broadcast to cluster subscribers via sharded fan-out. """ # Step 1: Channel Permissions & Rate Limits (Includes Persistent Mute Check) allowed, error_reason, cooldown_ms = self.channel_manager.validate_send_permission(request) if not allowed: return SendChatResponseDTO( success=False, error_message=error_reason, cooldown_remaining_ms=cooldown_ms ) # Step 2: Content Moderation & Anti-RMT Sentinel mod_result = self.moderator.process(request.content) if mod_result.should_auto_mute: # Persist mute state so subsequent messages are blocked at Step 1 self.channel_manager.mute_player(request.sender_id, duration_seconds=1800.0) return SendChatResponseDTO( success=False, error_message="Tài khoản bị tạm khóa chat do vi phạm chính sách an ninh (RMT / Lừa đảo / Quảng cáo)!", risk_score=mod_result.risk_score ) # Step 3: Linked Items verification linked_snapshots: List[ItemSnapshotDTO] = [] for item_uuid_str in request.link_item_uuids: snap_entry = self.item_link_service._snapshot_cache.get(item_uuid_str) if snap_entry: snapshot, _ = snap_entry linked_snapshots.append(snapshot) # Step 4: Construct ChatMessageDTO msg_id = f"msg_{int(time.time() * 1000)}_{uuid.uuid4().hex[:8]}" now_ms = int(time.time() * 1000) message = ChatMessageDTO( message_id=msg_id, sender_id=request.sender_id, sender_name=request.sender_name, sender_title=request.sender_title, sender_guild_tag=request.sender_guild_tag, channel=request.channel, target_id=request.target_id, raw_content=request.content, filtered_content=mod_result.filtered_content, timestamp_ms=now_ms, is_censored=mod_result.is_censored, moderation_risk_score=mod_result.risk_score, linked_items=tuple(linked_snapshots), zone_id=request.zone_id, guild_id=request.guild_id, party_id=request.party_id, ) # Step 5: Record send time and update channel history self.channel_manager.record_message_sent(request) channel_key = self.channel_manager.get_channel_key( channel=request.channel, zone_id=request.zone_id, guild_id=request.guild_id, party_id=request.party_id, player_a=request.sender_id, player_b=request.target_id ) self.channel_manager.add_to_history(channel_key, message) # Step 6: Asynchronous Non-blocking Broadcast await self.cluster_router.broadcast_to_channel(channel_key, message) return SendChatResponseDTO( success=True, message_id=msg_id, filtered_content=mod_result.filtered_content, risk_score=mod_result.risk_score ) def mute_player(self, player_id: int, duration_seconds: float = 1800.0) -> None: """Applies a persistent mute lock on a player account.""" self.channel_manager.mute_player(player_id, duration_seconds) def unmute_player(self, player_id: int) -> None: """Lifts a chat mute restriction from a player account.""" self.channel_manager.unmute_player(player_id) def is_player_muted(self, player_id: int) -> Tuple[bool, int]: """Checks if a player is currently muted, returning (is_muted, remaining_seconds).""" return self.channel_manager.is_player_muted(player_id) def query_item_snapshot(self, item_uuid: str, claimed_signature: str) -> QueryItemSnapshotResponseDTO: """Retrieves and cryptographically validates an item snapshot.""" return self.item_link_service.query_item_snapshot(item_uuid, claimed_signature) def subscribe_client(self, channel_key: str, client_id: int, callback: SubscriberCallback) -> None: """Subscribes a client socket / session to a channel key.""" self.cluster_router.registry.subscribe(channel_key, client_id, callback) def unsubscribe_client(self, channel_key: str, client_id: int) -> None: """Unsubscribes a client socket / session from a channel key.""" self.cluster_router.registry.unsubscribe(channel_key, client_id) def get_channel_history(self, channel_key: str) -> List[ChatMessageDTO]: """Fetches recent buffered messages for a channel.""" return self.channel_manager.get_history(channel_key) def clear_channel_history(self, channel_key: str) -> None: """Clears buffered messages for a channel.""" self.channel_manager.clear_history(channel_key) def reset_rate_limits(self) -> None: """Resets rate limiting state for testing and cluster maintenance.""" self.channel_manager.reset_rate_limits() def set_blacklist_entry(self, player_id: int, blocked_player_id: int, is_blocked: bool) -> None: """Manages player blacklists for Whisper spam protection.""" self.channel_manager.set_blacklist_entry(player_id, blocked_player_id, is_blocked)