""" Multi-Channel Policy, Rate Limiting, and Channel History Manager. Enforces channel-specific gates (Level requirements, Cooldown token buckets, Anti-spam). """ from __future__ import annotations import time from collections import deque from typing import Deque, Dict, List, Optional, Set, Tuple from server.chat.chat_types import ( ChatChannelType, ChatMessageDTO, SendChatRequestDTO, ) class ChannelManager: """ Manages channel routing rules, anti-spam rate limits, historical buffers, and persistent mute security states across all 8 chat channels. """ MAX_HISTORY_PER_CHANNEL = 100 MIN_WORLD_LEVEL = 20 MIN_RECRUIT_LEVEL = 10 COOLDOWNS: Dict[ChatChannelType, float] = { ChatChannelType.WORLD: 15.0, # 15s cooldown ChatChannelType.ZONE: 3.0, # 3s cooldown ChatChannelType.RECRUIT: 10.0, # 10s cooldown ChatChannelType.GUILD: 0.5, # 0.5s anti-flood ChatChannelType.PARTY: 0.2, # Ultra-low latency team channel ChatChannelType.WHISPER: 0.5, # 0.5s direct message cooldown ChatChannelType.FEEDBACK: 5.0, # 5s GM ticket cooldown ChatChannelType.SYSTEM: 0.0, # 0s authoritative announcement } def __init__(self) -> None: # (player_id, channel) -> last_timestamp self._last_send_time: Dict[Tuple[int, ChatChannelType], float] = {} # (player_id, channel) -> (last_message_content, timestamp) self._last_message_info: Dict[Tuple[int, ChatChannelType], Tuple[str, float]] = {} # player_id -> set of blocked player_ids (for Whisper) self._player_blacklists: Dict[int, Set[int]] = {} # channel_key -> deque of recent ChatMessageDTO self._channel_histories: Dict[str, Deque[ChatMessageDTO]] = {} # player_id -> mute_until_timestamp self._muted_players: Dict[int, float] = {} def get_channel_key( self, channel: ChatChannelType, zone_id: str = "", guild_id: str = "", party_id: str = "", player_a: int = 0, player_b: int = 0 ) -> str: """Derives a deterministic partition key for a channel context.""" if channel == ChatChannelType.WORLD: return "world" if channel == ChatChannelType.ZONE: return f"zone:{zone_id}" if channel == ChatChannelType.GUILD: return f"guild:{guild_id}" if channel == ChatChannelType.PARTY: return f"party:{party_id}" if channel == ChatChannelType.WHISPER: pair = sorted([player_a, player_b]) return f"whisper:{pair[0]}:{pair[1]}" if channel == ChatChannelType.RECRUIT: return "recruit" if channel == ChatChannelType.FEEDBACK: return f"feedback:{player_a}" if channel == ChatChannelType.SYSTEM: return "system" return "unspecified" def mute_player(self, player_id: int, duration_seconds: float = 1800.0) -> None: """Applies a persistent mute lock on a player account.""" self._muted_players[player_id] = time.time() + duration_seconds def unmute_player(self, player_id: int) -> None: """Lifts chat restrictions from a player account.""" self._muted_players.pop(player_id, None) def is_player_muted(self, player_id: int) -> Tuple[bool, int]: """Checks if a player is currently muted, returning (is_muted, remaining_seconds).""" mute_until = self._muted_players.get(player_id) if mute_until is None: return False, 0 now = time.time() if now < mute_until: return True, int(mute_until - now) # Mute expired del self._muted_players[player_id] return False, 0 def validate_send_permission( self, request: SendChatRequestDTO ) -> Tuple[bool, str, int]: """ Validates whether the player meets channel constraints. Returns: (is_allowed, error_reason, cooldown_remaining_ms) """ # 0. Persistent Mute Check is_muted, remaining_sec = self.is_player_muted(request.sender_id) if is_muted: return False, f"Tài khoản đang bị tạm khóa chat. Thời gian còn lại: {remaining_sec}s.", remaining_sec * 1000 now = time.time() channel = request.channel # 1. Level Gates if channel == ChatChannelType.WORLD and request.sender_level < self.MIN_WORLD_LEVEL: return False, f"Cần đạt cấp {self.MIN_WORLD_LEVEL} để phát tán kênh Thế Giới.", 0 if channel == ChatChannelType.RECRUIT and request.sender_level < self.MIN_RECRUIT_LEVEL: return False, f"Cần đạt cấp {self.MIN_RECRUIT_LEVEL} để dùng kênh Chiêu Mộ.", 0 # 2. Scope identifiers if channel == ChatChannelType.GUILD and not request.guild_id: return False, "Chưa gia nhập Bang Hội.", 0 if channel == ChatChannelType.PARTY and not request.party_id: return False, "Chưa gia nhập Tổ Đội.", 0 if channel == ChatChannelType.WHISPER: if request.target_id <= 0: return False, "Mục tiêu mật thư không hợp lệ.", 0 if request.target_id == request.sender_id: return False, "Không thể tự gửi mật thư cho chính mình.", 0 # Check target blacklist target_blacklist = self._player_blacklists.get(request.target_id, set()) if request.sender_id in target_blacklist: return False, "Đối phương đã chặn tin nhắn từ bạn.", 0 # 3. Cooldown & Token Bucket cooldown_duration = self.COOLDOWNS.get(channel, 1.0) last_time = self._last_send_time.get((request.sender_id, channel), 0.0) elapsed = now - last_time if elapsed < cooldown_duration: remaining_ms = int((cooldown_duration - elapsed) * 1000) return False, f"Gửi tin quá nhanh! Vui lòng đợi {remaining_ms}ms.", remaining_ms # 4. Duplicate spam detection (same exact message sent within 5 seconds on World/Zone) if channel in (ChatChannelType.WORLD, ChatChannelType.ZONE): last_info = self._last_message_info.get((request.sender_id, channel)) if last_info is not None: last_content, last_msg_time = last_info time_since_last = now - last_msg_time if request.content.strip().lower() == last_content.strip().lower() and time_since_last < 5.0: remaining_spam_ms = int((5.0 - time_since_last) * 1000) return False, "Phát hiện spam tin nhắn trùng lặp!", remaining_spam_ms return True, "", 0 def record_message_sent(self, request: SendChatRequestDTO) -> None: """Updates player cooldown and spam trackers upon message acceptance.""" now = time.time() self._last_send_time[(request.sender_id, request.channel)] = now self._last_message_info[(request.sender_id, request.channel)] = (request.content, now) def reset_rate_limits(self) -> None: """Clears transient rate-limiting and duplicate message tracking records.""" self._last_send_time.clear() self._last_message_info.clear() def add_to_history(self, channel_key: str, message: ChatMessageDTO) -> None: """Stores message into recent channel history ring buffer for broadcast channels.""" if channel_key.startswith("whisper:") or channel_key.startswith("feedback:"): return if channel_key not in self._channel_histories: self._channel_histories[channel_key] = deque(maxlen=self.MAX_HISTORY_PER_CHANNEL) self._channel_histories[channel_key].append(message) def get_history(self, channel_key: str) -> List[ChatMessageDTO]: """Retrieves recent messages from a channel.""" history = self._channel_histories.get(channel_key) return list(history) if history else [] def clear_history(self, channel_key: str) -> None: """Clears the history ring buffer for a channel.""" self._channel_histories.pop(channel_key, None) def set_blacklist_entry(self, player_id: int, blocked_player_id: int, is_blocked: bool) -> None: """Adds or removes a player from a blacklist.""" if player_id not in self._player_blacklists: self._player_blacklists[player_id] = set() if is_blocked: self._player_blacklists[player_id].add(blocked_player_id) else: self._player_blacklists[player_id].discard(blocked_player_id)