.gitea/workflows/ci-cd.yml

@@ -70,6 +70,9 @@ jobs:
           PanelURL = https://panel.example.com
           ClientAPIKey = ptlc_test_client_key_123456789
           ApplicationAPIKey = ptla_test_app_key_987654321
+          # Pelican Panel test keys
+          # ClientAPIKey = pacc_test_client_key_123456789
+          # ApplicationAPIKey = papp_test_app_key_987654321
 
           [Discord]
           Token = test_discord_token_placeholder
@@ -81,8 +84,13 @@ jobs:
           python -m pytest test_pterodisbot.py \
             -v \
             --tb=short \
-            --cov=pterodisbot \
-            --cov=server_metrics_graphs \
+            --cov=config \
+            --cov=api \
+            --cov=embeds \
+            --cov=views \
+            --cov=bot \
+            --cov=metrics \
+            --cov=models \
             --cov-report=xml \
             --cov-report=term \
             --cov-report=html \
@@ -129,7 +137,7 @@ jobs:
 
       - name: Run flake8
         run: |
-          flake8 pterodisbot.py server_metrics_graphs.py \
+          flake8 config.py api.py embeds.py views.py bot.py metrics.py models.py \
             --max-line-length=140 \
             --ignore=E501,W503,E203 \
             --exclude=venv,__pycache__,build,dist \
@@ -139,7 +147,7 @@ jobs:
 
       - name: Run pylint
         run: |
-          pylint pterodisbot.py server_metrics_graphs.py \
+          pylint config.py api.py embeds.py views.py bot.py metrics.py models.py \
             --disable=C0111,C0103,R0913,R0914,R0915,W0718 \
             --max-line-length=140 \
             --output-format=text \
@@ -148,17 +156,17 @@ jobs:
 
       - name: Check code formatting with black
         run: |
-          black --check --line-length=140 --diff pterodisbot.py server_metrics_graphs.py | tee black-report.txt
+          black --check --line-length=140 --diff config.py api.py embeds.py views.py bot.py metrics.py models.py | tee black-report.txt
         continue-on-error: true
 
       - name: Check import ordering
         run: |
-          isort --check-only --profile black --line-length=140 pterodisbot.py server_metrics_graphs.py
+          isort --check-only --profile black --line-length=140 config.py api.py embeds.py views.py bot.py metrics.py models.py
         continue-on-error: true
 
       - name: Type checking with mypy
         run: |
-          mypy pterodisbot.py server_metrics_graphs.py --ignore-missing-imports > mypy-report.txt || true
+          mypy config.py api.py embeds.py views.py bot.py metrics.py models.py --ignore-missing-imports > mypy-report.txt || true
         continue-on-error: true
 
       - name: Upload linting reports
@@ -203,7 +211,7 @@ jobs:
 
       - name: Run bandit security scan
         run: |
-          bandit -r . \
+          bandit -r config.py api.py embeds.py views.py bot.py metrics.py models.py \
             -f json \
             -o bandit-report.json \
             -ll \
@@ -273,6 +281,9 @@ jobs:
           PanelURL = https://panel.example.com
           ClientAPIKey = ptlc_test_client_key_123456789
           ApplicationAPIKey = ptla_test_app_key_987654321
+          # Pelican Panel test keys
+          # ClientAPIKey = pacc_test_client_key_123456789
+          # ApplicationAPIKey = papp_test_app_key_987654321
 
           [Discord]
           Token = test_discord_token_placeholder

__init__.py

@@ -0,0 +1,44 @@
+"""
+Pterodactyl Discord Bot
+
+A feature-rich Discord bot for monitoring and managing Pterodactyl/Pelican Panel
+game servers through Discord, providing real-time status monitoring and control
+capabilities with an intuitive interface.
+
+Modules:
+- config: Configuration management and validation
+- api: Panel API client (Pterodactyl/Pelican)
+- models: Pydantic data models
+- metrics: Server metrics tracking and graph generation
+- embeds: Discord embed management and persistence
+- views: Discord UI components (buttons, dropdowns)
+- bot: Main bot class and command definitions
+- exceptions: Custom exception classes
+"""
+
+__version__ = "2.0.0"
+__author__ = "Eaven Kimura"
+
+from config import Config, load_config, generate_config_from_env
+from exceptions import (
+    PterodisBotError,
+    ConfigValidationError,
+    APIError,
+    ServerNotFoundError,
+    PermissionError,
+    EmbedNotFoundError,
+    MetricsError,
+)
+
+__all__ = [
+    "Config",
+    "load_config",
+    "generate_config_from_env",
+    "PterodisBotError",
+    "ConfigValidationError",
+    "APIError",
+    "ServerNotFoundError",
+    "PermissionError",
+    "EmbedNotFoundError",
+    "MetricsError",
+]

api.py

@@ -0,0 +1,341 @@
+"""
+Pterodactyl/Pelican Panel API Client Module
+
+Provides a clean interface for interacting with Pterodactyl and Pelican Panel APIs.
+Supports both client API (server operations) and application API (admin operations).
+"""
+
+import asyncio
+import logging
+from typing import Any, Dict, List, Optional
+
+import aiohttp
+
+from config import Config
+from exceptions import APIError, ServerNotFoundError
+
+logger = logging.getLogger("pterodisbot")
+
+
+class PanelAPI:
+    """
+    Unified API client for Pterodactyl and Pelican Panel.
+
+    Handles all HTTP requests to the panel API with proper authentication,
+    error handling, and rate limiting protection.
+
+    Features:
+    - Automatic session management
+    - Dual API key support (client + application)
+    - Concurrent request protection via asyncio.Lock
+    - Comprehensive error handling and logging
+    - Response parsing with structured error information
+
+    Args:
+        config: Validated Config instance containing panel credentials
+    """
+
+    def __init__(self, config: Config) -> None:
+        """
+        Initialize the Panel API client.
+
+        Args:
+            config: Validated Config instance with panel URL and API keys
+        """
+        self.config = config
+        self.panel_url = config.pterodactyl_url.rstrip("/")
+        self.client_api_key = config.client_api_key
+        self.application_api_key = config.application_api_key
+        self.session: Optional[aiohttp.ClientSession] = None
+        self._lock = asyncio.Lock()
+        self._is_initialized = False
+        logger.info(f"Initialized {config.get_panel_display_name()} API client")
+
+    async def initialize(self) -> None:
+        """Initialize the aiohttp client session."""
+        if self.session is None or self.session.closed:
+            self.session = aiohttp.ClientSession()
+            logger.debug("Created new aiohttp ClientSession")
+        self._is_initialized = True
+
+    async def close(self) -> None:
+        """Cleanly close the aiohttp session."""
+        if self.session and not self.session.closed:
+            await self.session.close()
+            logger.debug("Closed aiohttp ClientSession")
+        self._is_initialized = False
+
+    async def _request(
+        self,
+        method: str,
+        endpoint: str,
+        data: Optional[Dict[str, Any]] = None,
+        use_application_key: bool = False,
+    ) -> Dict[str, Any]:
+        """
+        Make an authenticated request to the panel API.
+
+        Args:
+            method: HTTP method (GET, POST, PUT, DELETE)
+            endpoint: API endpoint (e.g., 'application/servers')
+            data: Optional JSON payload for POST/PUT requests
+            use_application_key: Whether to use application API key (admin endpoints)
+
+        Returns:
+            Dictionary containing API response data
+
+        Raises:
+            APIError: For network errors, HTTP errors, or parsing failures
+        """
+        if not self._is_initialized or self.session is None or self.session.closed:
+            await self.initialize()
+
+        url = f"{self.panel_url}/api/{endpoint}"
+        api_key_type = "Application" if use_application_key else "Client"
+        logger.debug(f"Preparing {method} request to {endpoint} using {api_key_type} API key")
+
+        api_key = self.application_api_key if use_application_key else self.client_api_key
+        headers = {
+            "Authorization": f"Bearer {api_key}",
+            "Accept": "application/json",
+            "Content-Type": "application/json",
+        }
+
+        try:
+            async with self._lock:
+                logger.debug(f"Acquired lock for API request to {endpoint}")
+                async with self.session.request(
+                    method, url, headers=headers, json=data
+                ) as response:
+                    if response.status == 204:
+                        logger.debug(f"Received 204 No Content from {endpoint}")
+                        return {"status": "success"}
+
+                    response_data = await response.json()
+                    logger.debug(
+                        f"Received response from {endpoint} with status {response.status}"
+                    )
+
+                    if response.status >= 400:
+                        error_msg = self._extract_error_message(response_data)
+                        logger.error(
+                            f"API request to {endpoint} failed with status "
+                            f"{response.status}: {error_msg}"
+                        )
+                        raise APIError(
+                            message=error_msg,
+                            endpoint=endpoint,
+                            status_code=response.status,
+                        )
+
+                    return response_data
+
+        except aiohttp.ClientError as e:
+            logger.error(f"Network error during API request to {endpoint}: {str(e)}")
+            raise APIError(
+                message=f"Network error: {str(e)}",
+                endpoint=endpoint,
+            )
+        except Exception as e:
+            logger.error(f"Unexpected error during API request to {endpoint}: {str(e)}")
+            raise APIError(
+                message=f"Unexpected error: {str(e)}",
+                endpoint=endpoint,
+            )
+
+    def _extract_error_message(self, response_data: Dict[str, Any]) -> str:
+        """
+        Extract human-readable error message from API response.
+
+        Args:
+            response_data: Parsed JSON response from API
+
+        Returns:
+            Formatted error message string
+        """
+        if "errors" in response_data and isinstance(response_data["errors"], list):
+            errors = response_data["errors"]
+            if errors and isinstance(errors[0], dict):
+                return errors[0].get("detail", "Unknown API error")
+            return str(errors[0]) if errors else "Unknown API error"
+        return response_data.get("message", "Unknown error")
+
+    # ============================================================
+    # Server Management Methods (Client API)
+    # ============================================================
+
+    async def get_server_resources(self, server_id: str) -> Dict[str, Any]:
+        """
+        Get resource usage for a specific server.
+
+        Args:
+            server_id: The panel server identifier
+
+        Returns:
+            Dictionary containing server resource usage and state
+
+        Raises:
+            ServerNotFoundError: If server doesn't exist
+            APIError: For other API errors
+        """
+        logger.debug(f"Fetching resource usage for server {server_id}")
+        try:
+            response = await self._request(
+                "GET", f"client/servers/{server_id}/resources"
+            )
+            return response
+        except APIError as e:
+            if e.status_code == 404:
+                raise ServerNotFoundError(f"Server {server_id} not found")
+            raise
+
+    async def send_power_action(self, server_id: str, action: str) -> Dict[str, Any]:
+        """
+        Send a power action to a server.
+
+        Args:
+            server_id: The panel server identifier
+            action: Power action (start, stop, restart)
+
+        Returns:
+            Dictionary containing API response status
+
+        Raises:
+            ValueError: If action is invalid
+            APIError: For API errors
+        """
+        valid_actions = ["start", "stop", "restart"]
+        if action not in valid_actions:
+            raise ValueError(
+                f"Invalid power action: {action}. Must be one of: {valid_actions}"
+            )
+
+        logger.info(f"Sending {action} command to server {server_id}")
+        result = await self._request(
+            "POST", f"client/servers/{server_id}/power", {"signal": action}
+        )
+
+        if result.get("status") == "success":
+            logger.info(f"Successfully executed {action} on server {server_id}")
+        else:
+            logger.error(f"Failed to execute {action} on server {server_id}")
+
+        return result
+
+    # ============================================================
+    # Server Information Methods (Application API)
+    # ============================================================
+
+    async def get_servers(self) -> List[Dict[str, Any]]:
+        """
+        Get a list of all servers from the panel.
+
+        Uses application API key as this is an admin endpoint.
+
+        Returns:
+            List of server dictionaries with full attributes
+        """
+        logger.info("Fetching list of all servers from panel")
+        response = await self._request("GET", "application/servers", use_application_key=True)
+        servers = response.get("data", [])
+        logger.info(f"Retrieved {len(servers)} servers from panel")
+        return servers
+
+    async def get_server_details(self, server_id: str) -> Dict[str, Any]:
+        """
+        Get detailed server information including allocations.
+
+        Uses application API key as this is an admin endpoint.
+
+        Args:
+            server_id: The panel server identifier
+
+        Returns:
+            Dictionary containing detailed server information
+        """
+        logger.debug(f"Fetching detailed information for server {server_id}")
+        return await self._request(
+            "GET", f"application/servers/{server_id}", use_application_key=True
+        )
+
+    async def get_server_allocations(self, server_id: str) -> Dict[str, Any]:
+        """
+        Get allocation information for a server (IP addresses and ports).
+
+        Uses application API key as this is an admin endpoint.
+
+        Args:
+            server_id: The panel server identifier
+
+        Returns:
+            Dictionary containing server allocation information
+        """
+        logger.debug(f"Fetching allocation information for server {server_id}")
+        return await self._request(
+            "GET",
+            f"application/servers/{server_id}/allocations",
+            use_application_key=True,
+        )
+
+    async def get_client_server_details(self, server_id: str) -> Dict[str, Any]:
+        """
+        Get server details using client API (for connection info).
+
+        Args:
+            server_id: The panel server identifier
+
+        Returns:
+            Dictionary containing server details with allocations
+        """
+        logger.debug(f"Fetching client server details for {server_id}")
+        return await self._request("GET", f"client/servers/{server_id}")
+
+    # ============================================================
+    # Convenience Methods
+    # ============================================================
+
+    async def get_server_default_allocation(
+        self, server_id: str
+    ) -> tuple[str, str]:
+        """
+        Get the default allocation IP and port for a server.
+
+        Args:
+            server_id: The panel server identifier
+
+        Returns:
+            Tuple of (ip_alias, port)
+
+        Raises:
+            ServerNotFoundError: If server or allocations not found
+            APIError: For API errors
+        """
+        details = await self.get_client_server_details(server_id)
+
+        if details.get("status") == "error":
+            raise ServerNotFoundError(
+                details.get("message", "Failed to get server details")
+            )
+
+        attributes = details.get("attributes", {})
+        relationships = attributes.get("relationships", {})
+        allocations = relationships.get("allocations", {}).get("data", [])
+
+        if not allocations:
+            raise ServerNotFoundError("No allocations found for this server")
+
+        # Find default allocation or fallback to first
+        default_allocation = next(
+            (
+                alloc
+                for alloc in allocations
+                if alloc.get("attributes", {}).get("is_default", False)
+            ),
+            allocations[0],
+        )
+
+        allocation_attrs = default_allocation.get("attributes", {})
+        ip_alias = allocation_attrs.get("ip_alias", "Unknown")
+        port = str(allocation_attrs.get("port", "Unknown"))
+
+        return ip_alias, port

bot.py

@@ -0,0 +1,707 @@
+"""
+Main Bot Module
+
+Core bot class and command definitions.
+Manages the bot lifecycle, background tasks, and Discord interactions.
+"""
+
+import asyncio
+import logging
+import signal
+import sys
+import types
+from datetime import datetime
+from typing import TYPE_CHECKING, Optional, Tuple
+
+import discord
+from discord.ext import commands, tasks
+
+from api import PanelAPI
+from config import Config
+from embeds import EmbedManager
+from exceptions import ConfigValidationError
+from metrics import ServerMetricsManager
+from models import ServerData, ServerResources
+
+if TYPE_CHECKING:
+    from views import ServerStatusView
+
+logger = logging.getLogger("pterodisbot")
+
+
+# Singleton instance holder
+_bot_instance: Optional["PterodactylBot"] = None
+
+
+def get_bot_instance() -> Optional["PterodactylBot"]:
+    """Get the singleton bot instance."""
+    return _bot_instance
+
+
+class PterodactylBot(commands.Bot):
+    """
+    Main bot class for Pterodactyl/Pelican Panel server management.
+
+    Handles Discord interactions, embed management, and background tasks.
+    Manages server status embeds and user commands.
+
+    Features:
+    - Background status update task with smart update detection
+    - Persistent embed tracking with JSON storage
+    - Real-time server monitoring with metrics graphs
+    - Role-based access control
+    - Single-guild restriction
+    - Graceful shutdown handling
+    """
+
+    def __init__(self, config: Config, *args, **kwargs) -> None:
+        """
+        Initialize the Pterodactyl bot instance.
+
+        Args:
+            config: Validated application configuration
+            *args: Additional arguments for commands.Bot
+            **kwargs: Additional keyword arguments for commands.Bot
+        """
+        global _bot_instance
+        _bot_instance = self
+
+        super().__init__(*args, **kwargs)
+
+        self.config = config
+        self.api = PanelAPI(config)
+        self.metrics_manager = ServerMetricsManager()
+        self.embed_manager = EmbedManager(config, self.api, self.metrics_manager)
+
+        # Server cache from panel
+        self.server_cache: dict = {}
+
+        # Concurrency control
+        self.update_lock = asyncio.Lock()
+
+        # State tracking for smart updates
+        # Format: {server_id: (state, cpu_usage, last_force_update)}
+        self.previous_states: dict = {}
+
+        logger.info("Initialized PterodactylBot instance with state tracking")
+
+    @classmethod
+    def get_instance(cls) -> Optional["PterodactylBot"]:
+        """Get the singleton bot instance."""
+        return _bot_instance
+
+    async def setup_hook(self) -> None:
+        """
+        Bot setup routine called when the bot is starting.
+
+        Initializes API client, loads saved data, and starts background tasks.
+        """
+        logger.info("Running bot setup hook")
+
+        # Initialize API client
+        await self.api.initialize()
+        logger.info(f"Initialized {self.config.get_panel_display_name()} API client")
+
+        # Load saved embed locations
+        await self.embed_manager.load_locations()
+
+        # Start background update task
+        self.update_status.start()
+        logger.info("Started background status update task")
+
+    @tasks.loop(seconds=10)  # UPDATE_INTERVAL = 10
+    async def update_status(self) -> None:
+        """
+        Background task to update server status embeds when:
+        1. Server power state changes (started/stopped/restarted)
+        2. Significant CPU usage change (>50% difference)
+        3. First time seeing the server
+        4. Server has been running for 10 minutes (force update for uptime)
+
+        This minimizes API calls to Discord and updates while maintaining
+        real-time awareness of important server changes.
+        """
+        logger.info("Starting optimized server status update cycle")
+        async with self.update_lock:
+            try:
+                # Fetch current server list from panel
+                servers = await self.api.get_servers()
+                if not servers:
+                    logger.warning("No servers found in panel during update")
+                    return
+
+                # Update our local cache with fresh server data
+                self.server_cache = {s["attributes"]["identifier"]: s for s in servers}
+                logger.debug(f"Updated server cache with {len(servers)} servers")
+
+                # Clean up metrics for servers that no longer exist
+                active_server_ids = list(self.server_cache.keys())
+                self.metrics_manager.cleanup_old_servers(active_server_ids)
+
+                # Update statistics
+                update_count = 0
+                error_count = 0
+                missing_count = 0
+                skipped_count = 0
+                current_time = datetime.now().timestamp()
+
+                # Process each server we're tracking embeds for
+                for server_id, location in list(self.embed_manager.embed_locations.items()):
+                    # Skip if server no longer exists in panel
+                    if server_id not in self.server_cache:
+                        logger.warning(f"Server {server_id} not found in cache, skipping update")
+                        continue
+
+                    server_data = self.server_cache[server_id]
+                    server_name = server_data["attributes"]["name"]
+
+                    try:
+                        logger.debug(f"Checking status for server {server_name} ({server_id})")
+
+                        # Get current server resource usage
+                        resources = await self.api.get_server_resources(server_id)
+                        current_state = resources["attributes"]["current_state"]
+                        cpu_usage = round(
+                            resources["attributes"]["resources"].get("cpu_absolute", 0), 2
+                        )
+
+                        # Collect metrics data for running servers
+                        if current_state == "running":
+                            memory_usage = round(
+                                resources["attributes"]["resources"].get("memory_bytes", 0)
+                                / (1024**2),
+                                2,
+                            )
+                            self.metrics_manager.add_server_data(
+                                server_id, server_name, cpu_usage, memory_usage
+                            )
+                            logger.debug(
+                                f"Added metrics data for {server_name}: "
+                                f"CPU={cpu_usage}%, Memory={memory_usage}MB"
+                            )
+
+                        # Retrieve previous recorded state
+                        prev_state, prev_cpu, last_force_update = self.previous_states.get(
+                            server_id, (None, 0, None)
+                        )
+
+                        # DECISION LOGIC: Should we update the embed?
+                        needs_update = False
+
+                        # 1. Check if power state changed (most important)
+                        if current_state != prev_state:
+                            logger.debug(
+                                f"Power state changed for {server_name}: "
+                                f"{prev_state} -> {current_state}"
+                            )
+                            needs_update = True
+
+                        # 2. Check for significant CPU change (only if server is running)
+                        elif current_state == "running" and abs(cpu_usage - prev_cpu) > 50:
+                            logger.debug(
+                                f"Significant CPU change for {server_name}: "
+                                f"{prev_cpu}% -> {cpu_usage}%"
+                            )
+                            needs_update = True
+
+                        # 3. First time we're seeing this server (initial update)
+                        elif prev_state is None:
+                            logger.debug(f"First check for {server_name}, performing initial update")
+                            needs_update = True
+
+                        # 4. Force update every 10 minutes for running servers (for uptime counter)
+                        elif current_state == "running" and (
+                            last_force_update is None
+                            or current_time - last_force_update >= 600
+                        ):  # 10 minutes = 600 seconds
+                            logger.debug(f"Executing 10-minute force update for running server {server_name}")
+                            needs_update = True
+                            last_force_update = current_time
+
+                        # PERFORM UPDATE IF NEEDED
+                        if needs_update:
+                            # Convert to models for embed creation
+                            server_obj = ServerData.model_validate(server_data)
+                            resources_obj = ServerResources.model_validate(resources)
+
+                            # Update the embed
+                            success = await self.embed_manager.update_single_embed(
+                                server_id=server_id,
+                                location=location,
+                                server_data=server_obj,
+                                resources=resources_obj,
+                                current_state=current_state,
+                                cpu_usage=cpu_usage,
+                            )
+
+                            if success:
+                                update_count += 1
+                                # Update state tracking
+                                new_last_force_update = (
+                                    last_force_update
+                                    if needs_update
+                                    and current_state == "running"
+                                    and current_time - (last_force_update or 0) >= 600
+                                    else last_force_update
+                                )
+                                self.previous_states[server_id] = (
+                                    current_state,
+                                    cpu_usage,
+                                    new_last_force_update,
+                                )
+                        else:
+                            # No significant changes, but update tracking with current state
+                            self.previous_states[server_id] = (
+                                current_state,
+                                cpu_usage,
+                                last_force_update,
+                            )
+                            skipped_count += 1
+                            logger.debug(f"No changes detected for {server_name}, skipping update")
+
+                    except discord.NotFound:
+                        # Embed message was deleted - clean up tracking
+                        logger.warning(f"Embed for server {server_id} not found, removing from tracking")
+                        await self.embed_manager.untrack_embed(server_id)
+                        missing_count += 1
+                    except Exception as e:
+                        logger.error(f"Failed to update status for server {server_id}: {str(e)}")
+                        error_count += 1
+
+                    # Small delay between servers to avoid rate limits
+                    await asyncio.sleep(0.5)
+
+                # Log summary of this update cycle
+                logger.info(
+                    f"Update cycle complete: "
+                    f"{update_count} updated, "
+                    f"{skipped_count} skipped, "
+                    f"{missing_count} missing, "
+                    f"{error_count} errors"
+                )
+
+            except Exception as e:
+                logger.error(f"Error in update_status task: {str(e)}")
+                await asyncio.sleep(5)
+
+    @update_status.before_loop
+    async def before_update_status(self) -> None:
+        """Wait for bot to be ready before starting update task."""
+        logger.debug("Waiting for bot readiness before starting update task")
+        await self.wait_until_ready()
+        await self.embed_manager.refresh_all_embeds()
+
+    @update_status.after_loop
+    async def after_update_status(self) -> None:
+        """Handle update task stopping."""
+        if self.update_status.is_being_cancelled():
+            logger.info("Server status update task was cancelled")
+        elif self.update_status.failed():
+            logger.error("Server status update task failed")
+
+    async def close(self) -> None:
+        """Cleanup when bot is shutting down."""
+        await self.embed_manager.save_locations()
+        if self.api:
+            await self.api.close()
+        await super().close()
+
+
+async def check_allowed_guild(interaction: discord.Interaction, config: Config) -> bool:
+    """
+    Verify that an interaction is coming from the allowed guild.
+
+    Args:
+        interaction: Discord interaction object
+        config: Application configuration
+
+    Returns:
+        bool: True if interaction is allowed, False otherwise
+    """
+    if interaction.guild_id != config.allowed_guild_id:
+        logger.warning(
+            f"Command attempted from unauthorized guild {interaction.guild_id} "
+            f"by {interaction.user.name}"
+        )
+        await interaction.response.send_message(
+            "This bot is only available in a specific server.", ephemeral=True
+        )
+        return False
+    return True
+
+
+async def setup_commands(bot: PterodactylBot) -> None:
+    """
+    Set up all slash commands and event handlers for the bot.
+
+    Args:
+        bot: The bot instance
+    """
+    config = bot.config
+
+    # ==============================================
+    # EVENT HANDLERS
+    # ==============================================
+
+    @bot.event
+    async def on_interaction(interaction: discord.Interaction) -> None:
+        """Global interaction handler to check guild before processing any interaction."""
+        if interaction.guild_id != config.allowed_guild_id:
+            logger.debug(f"Ignoring interaction from unauthorized guild {interaction.guild_id}")
+            await interaction.response.send_message(
+                "This bot is only available in a specific server.", ephemeral=True
+            )
+            return
+
+    @bot.event
+    async def on_ready() -> None:
+        """Called when the bot successfully connects to Discord."""
+        logger.info(f"Bot connected as {bot.user.name} (ID: {bot.user.id})")
+
+        try:
+            # Sync commands only to the allowed guild
+            guild = discord.Object(id=bot.config.allowed_guild_id)
+            bot.tree.copy_global_to(guild=guild)
+            synced = await bot.tree.sync(guild=guild)
+            logger.info(
+                f"Successfully synced {len(synced)} command(s) to guild {bot.config.allowed_guild_id}: "
+                f"{[cmd.name for cmd in synced]}"
+            )
+        except Exception as e:
+            logger.error(f"Command sync failed: {str(e)}")
+
+    # ==============================================
+    # SLASH COMMANDS
+    # ==============================================
+
+    @bot.tree.command(
+        name="server_status", description="Get a list of available game servers to control"
+    )
+    async def server_status(interaction: discord.Interaction) -> None:
+        """
+        Slash command to display server status dashboard with interactive dropdown selection.
+
+        Workflow:
+        1. Validates guild permissions and defers ephemeral response
+        2. Refreshes server cache from Panel API
+        3. Calculates online/offline statistics by checking each server's state
+        4. Creates statistics embed with visual server status breakdown
+        5. Generates dropdown menu with all servers (name + description)
+        6. Handles user selection via ephemeral dropdown interaction
+        7. Creates permanent status embed in channel upon selection
+        8. Manages embed tracking and cleanup of previous embeds
+        """
+        if not await check_allowed_guild(interaction, config):
+            return
+
+        logger.info(f"Server status command invoked by {interaction.user.name}")
+        await interaction.response.defer(ephemeral=True)
+
+        try:
+            # Refresh server cache with current data from panel
+            servers = await bot.api.get_servers()
+            if not servers:
+                logger.warning("No servers found in panel")
+                await interaction.followup.send(
+                    "No servers found in the panel.", ephemeral=True
+                )
+                return
+
+            bot.server_cache = {s["attributes"]["identifier"]: s for s in servers}
+            logger.debug(f"Refreshed server cache with {len(servers)} servers")
+
+            # Count online/offline servers
+            online_count = 0
+            offline_count = 0
+
+            for server_id, server_data in bot.server_cache.items():
+                resources = await bot.api.get_server_resources(server_id)
+                current_state = resources.get("attributes", {}).get("current_state", "offline")
+
+                if current_state == "running":
+                    online_count += 1
+                else:
+                    offline_count += 1
+
+            # Create statistics embed
+            stats_embed = discord.Embed(
+                title="🏗️ Server Status Dashboard",
+                description="Select a server from the dropdown below to view its detailed status and controls.",
+                color=discord.Color.blue(),
+                timestamp=datetime.now(),
+            )
+
+            stats_embed.add_field(
+                name="📊 Server Statistics",
+                value=f"**Total Servers:** {len(servers)}\n"
+                f"✅ **Online:** {online_count}\n"
+                f"❌ **Offline:** {offline_count}",
+                inline=False,
+            )
+
+            stats_embed.add_field(
+                name="ℹ️ How to Use",
+                value="Use the dropdown menu below to select a server. "
+                "The server's status embed will be posted in this channel.",
+                inline=False,
+            )
+
+            stats_embed.set_footer(text="Server status will update automatically")
+
+            # Create dropdown menu options
+            from views import ServerDropdown, ServerDropdownView
+
+            server_options = []
+            for server_id, server_data in bot.server_cache.items():
+                server_name = server_data["attributes"]["name"]
+                server_description = server_data["attributes"].get("description", "No description")
+
+                # Truncate description if too long for dropdown constraints
+                if len(server_description) > 50:
+                    server_description = server_description[:47] + "..."
+
+                server_options.append(
+                    discord.SelectOption(
+                        label=server_name, value=server_id, description=server_description
+                    )
+                )
+
+            # Send the initial dashboard embed with dropdown (ephemeral)
+            await interaction.followup.send(
+                embed=stats_embed,
+                view=ServerDropdownView(server_options),
+                ephemeral=True,
+            )
+            logger.info(f"Sent server status dashboard to {interaction.user.name} with {len(server_options)} servers")
+
+        except Exception as e:
+            logger.error(f"Server status command failed: {str(e)}")
+            await interaction.followup.send(
+                f"❌ Failed to load server status: {str(e)}", ephemeral=True
+            )
+
+    @bot.tree.command(
+        name="refresh_embeds", description="Refresh all server status embeds (admin only)"
+    )
+    async def refresh_embeds(interaction: discord.Interaction) -> None:
+        """Slash command to refresh all server embeds."""
+        if not await check_allowed_guild(interaction, config):
+            return
+
+        logger.info(f"Refresh embeds command invoked by {interaction.user.name}")
+        await interaction.response.defer(ephemeral=True)
+
+        # Require administrator permissions
+        if not interaction.user.guild_permissions.administrator:
+            logger.warning(f"Unauthorized refresh attempt by {interaction.user.name}")
+            await interaction.followup.send(
+                "You need administrator permissions to refresh all embeds.", ephemeral=True
+            )
+            return
+
+        try:
+            logger.info("Starting full embed refresh per admin request")
+            deleted, created = await bot.embed_manager.refresh_all_embeds()
+            await interaction.followup.send(
+                f"Refreshed all embeds. Deleted {deleted} old embeds, created {created} new ones.",
+                ephemeral=True,
+            )
+            logger.info(f"Embed refresh completed: {deleted} deleted, {created} created")
+        except Exception as e:
+            logger.error(f"Embed refresh failed: {str(e)}")
+            await interaction.followup.send(f"Failed to refresh embeds: {str(e)}", ephemeral=True)
+
+    @bot.tree.command(
+        name="purge_embeds",
+        description="Permanently delete all server status embeds (admin only)",
+    )
+    async def purge_embeds(interaction: discord.Interaction) -> None:
+        """
+        Slash command to permanently purge all server status embeds from Discord channels.
+
+        Workflow:
+        1. Validates administrator permissions
+        2. Checks if any embeds are currently tracked
+        3. Sends initial progress embed
+        4. Processes each embed sequentially with error handling
+        5. Updates progress embed in real-time
+        6. Saves cleared tracking data to disk
+        7. Sends final results with comprehensive statistics
+        """
+        if not await check_allowed_guild(interaction, config):
+            return
+
+        logger.info(f"Purge embeds command invoked by {interaction.user.name}")
+        await interaction.response.defer(ephemeral=True)
+
+        # Require administrator permissions
+        if not interaction.user.guild_permissions.administrator:
+            logger.warning(f"Unauthorized purge attempt by {interaction.user.name}")
+            await interaction.followup.send(
+                "You need administrator permissions to purge all embeds.", ephemeral=True
+            )
+            return
+
+        try:
+            logger.info("Starting embed purge per admin request")
+
+            deleted_count = 0
+            not_found_count = 0
+            error_count = 0
+            total_embeds = len(bot.embed_manager.embed_locations)
+
+            if total_embeds == 0:
+                await interaction.followup.send(
+                    "No embeds are currently being tracked. Nothing to purge.", ephemeral=True
+                )
+                return
+
+            # Create progress embed
+            progress_embed = discord.Embed(
+                title="🔄 Purging Server Embeds",
+                description=f"Processing {total_embeds} embeds...",
+                color=discord.Color.orange(),
+            )
+            progress_embed.add_field(name="Deleted", value="0", inline=True)
+            progress_embed.add_field(name="Not Found", value="0", inline=True)
+            progress_embed.add_field(name="Errors", value="0", inline=True)
+            progress_embed.set_footer(text="This may take a while...")
+
+            progress_message = await interaction.followup.send(embed=progress_embed, ephemeral=True)
+
+            # Process each tracked embed
+            for server_id, location in list(bot.embed_manager.embed_locations.items()):
+                try:
+                    channel = bot.get_channel(int(location["channel_id"]))
+                    if channel:
+                        try:
+                            message = await channel.fetch_message(int(location["message_id"]))
+                            await message.delete()
+                            deleted_count += 1
+                            logger.debug(f"Successfully purged embed for server {server_id}")
+                        except discord.NotFound:
+                            not_found_count += 1
+                            logger.debug(f"Embed for server {server_id} already deleted")
+                        except discord.Forbidden:
+                            error_count += 1
+                            logger.error(f"Permission denied when deleting embed for server {server_id}")
+                        except Exception as e:
+                            error_count += 1
+                            logger.error(f"Error deleting embed for server {server_id}: {str(e)}")
+                    else:
+                        not_found_count += 1
+                        logger.warning(f"Channel not found for server {server_id}")
+
+                    # Remove from tracking immediately
+                    await bot.embed_manager.untrack_embed(server_id)
+
+                    # Update progress every 5 embeds or for the last one
+                    processed = deleted_count + not_found_count + error_count
+                    if processed % 5 == 0 or processed == total_embeds:
+                        progress_embed.description = f"Processed {processed}/{total_embeds} embeds"
+                        progress_embed.set_field_at(0, name="Deleted", value=str(deleted_count), inline=True)
+                        progress_embed.set_field_at(1, name="Not Found", value=str(not_found_count), inline=True)
+                        progress_embed.set_field_at(2, name="Errors", value=str(error_count), inline=True)
+
+                        await progress_message.edit(embed=progress_embed)
+
+                    # Small delay to avoid rate limits
+                    await asyncio.sleep(0.3)
+
+                except Exception as e:
+                    error_count += 1
+                    logger.error(f"Unexpected error processing server {server_id}: {str(e)}")
+
+            # Create results embed
+            result_embed = discord.Embed(
+                title="✅ Embed Purge Complete",
+                color=discord.Color.green(),
+                timestamp=datetime.now(),
+            )
+            result_embed.add_field(name="Total Tracked", value=str(total_embeds), inline=True)
+            result_embed.add_field(name="✅ Successfully Deleted", value=str(deleted_count), inline=True)
+            result_embed.add_field(name="❌ Already Missing", value=str(not_found_count), inline=True)
+            result_embed.add_field(name="⚠️ Errors", value=str(error_count), inline=True)
+            result_embed.add_field(
+                name="📊 Success Rate",
+                value=f"{((deleted_count + not_found_count) / total_embeds * 100):.1f}%",
+                inline=True,
+            )
+            result_embed.set_footer(text="Embed tracking file has been cleared")
+
+            await progress_message.edit(embed=result_embed)
+            logger.info(f"Embed purge completed: {deleted_count} deleted, {not_found_count} not found, {error_count} errors")
+
+        except Exception as e:
+            logger.error(f"Embed purge failed: {str(e)}")
+            await interaction.followup.send(f"❌ Failed to purge embeds: {str(e)}", ephemeral=True)
+
+
+# ==============================================
+# SYSTEM SIGNAL HANDLERS
+# ==============================================
+
+
+def handle_sigint(signum: int, frame: types.FrameType) -> None:
+    """
+    Handle SIGINT signals (Ctrl+C) by initiating graceful shutdown.
+
+    Args:
+        signum: The signal number (signal.SIGINT)
+        frame: Current stack frame (unused but required by signal handler signature)
+    """
+    logger.info("Received SIGINT (Ctrl+C), initiating graceful shutdown...")
+    raise KeyboardInterrupt
+
+
+def handle_sigterm(signum: int, frame: types.FrameType) -> None:
+    """
+    Handle SIGTERM signals (container stop) by initiating graceful shutdown.
+
+    Args:
+        signum: The signal number (signal.SIGTERM)
+        frame: Current stack frame (unused but required by signal handler signature)
+    """
+    logger.info("Received SIGTERM (container stop), initiating graceful shutdown...")
+    raise KeyboardInterrupt
+
+
+# ==============================================
+# BOT STARTUP
+# ==============================================
+
+
+async def run_bot(config: Config) -> None:
+    """
+    Main async entry point for the bot application.
+
+    Args:
+        config: Validated application configuration
+    """
+    logger.info("Starting bot initialization")
+
+    # Create bot instance
+    intents = discord.Intents.default()
+    intents.message_content = True
+
+    bot = PterodactylBot(config, command_prefix="!", intents=intents)
+
+    # Set up commands and events
+    await setup_commands(bot)
+
+    # Register signal handlers
+    signal.signal(signal.SIGINT, handle_sigint)
+    signal.signal(signal.SIGTERM, handle_sigterm)
+    logger.info("System signal handlers registered")
+
+    try:
+        await bot.start(config.discord_token)
+    except KeyboardInterrupt:
+        logger.info("Bot shutting down")
+    except Exception as e:
+        logger.error(f"Bot crashed: {str(e)}")
+        sys.exit(1)
+    finally:
+        logger.info("Bot shutdown initiated - performing cleanup")
+        await bot.close()
+        logger.info("Bot shutdown complete")

config.py

@@ -0,0 +1,207 @@
+"""
+Configuration Management Module
+
+Handles loading, validation, and access to application configuration.
+Supports both config.ini files and environment variable overrides.
+"""
+
+import configparser
+import logging
+import os
+from pathlib import Path
+from typing import Optional
+
+from exceptions import ConfigValidationError
+
+logger = logging.getLogger("pterodisbot")
+
+
+class Config:
+    """
+    Application configuration manager.
+
+    Loads configuration from config.ini file, validates all required values,
+    and provides typed access to configuration parameters.
+    Supports both Pterodactyl (ptlc_/ptla_) and Pelican Panel (pacc_/papp_)
+    API key formats.
+
+    Attributes:
+        pterodactyl_url: URL of the panel (must include http:// or https://)
+        client_api_key: Client API key for server operations
+        application_api_key: Application API key for admin operations
+        discord_token: Discord bot token
+        allowed_guild_id: Discord server ID where bot operates
+        required_role: Role name required for server control permissions
+        update_interval: Background task update interval in seconds
+        embed_storage_path: Path to embed locations JSON file
+    """
+
+    def __init__(self, config_path: str = "config.ini") -> None:
+        """
+        Initialize configuration from file.
+
+        Args:
+            config_path: Path to config.ini file
+
+        Raises:
+            ConfigValidationError: If required values are missing or invalid
+        """
+        self._config = configparser.ConfigParser()
+        self._config_path = Path(config_path)
+
+        if not self._config_path.exists():
+            raise ConfigValidationError(f"Configuration file not found: {config_path}")
+
+        self._config.read(self._config_path)
+        self._validate()
+
+        # Validated configuration values
+        self.pterodactyl_url: str = self._config.get("Pterodactyl", "PanelURL").rstrip("/")
+        self.client_api_key: str = self._config.get("Pterodactyl", "ClientAPIKey")
+        self.application_api_key: str = self._config.get("Pterodactyl", "ApplicationAPIKey")
+        self.discord_token: str = self._config.get("Discord", "Token")
+        self.allowed_guild_id: int = self._config.getint("Discord", "AllowedGuildID")
+        self.required_role: str = "Game Server User"
+        self.update_interval: int = 10
+        self.embed_storage_path: Path = Path("./embed/embed_locations.json")
+
+        logger.info("Configuration loaded and validated successfully")
+
+    def _validate(self) -> None:
+        """
+        Validate all required configuration values.
+
+        Checks:
+        - Required sections exist
+        - All required keys are present
+        - AllowedGuildID is a positive integer
+        - API keys have correct prefixes (ptlc_/pacc_ and ptla_/papp_)
+        - PanelURL has valid protocol
+
+        Raises:
+            ConfigValidationError: If any validation fails
+        """
+        errors = []
+
+        # Validate Pterodactyl section
+        if not self._config.has_section("Pterodactyl"):
+            errors.append("Missing [Pterodactyl] section in config.ini")
+        else:
+            required_ptero = ["PanelURL", "ClientAPIKey", "ApplicationAPIKey"]
+            for key in required_ptero:
+                if not self._config.get("Pterodactyl", key, fallback=None):
+                    errors.append(f"Missing required Pterodactyl config value: {key}")
+
+        # Validate Discord section
+        if not self._config.has_section("Discord"):
+            errors.append("Missing [Discord] section in config.ini")
+        else:
+            required_discord = ["Token", "AllowedGuildID"]
+            for key in required_discord:
+                if not self._config.get("Discord", key, fallback=None):
+                    errors.append(f"Missing required Discord config value: {key}")
+
+        # Validate AllowedGuildID is a valid integer
+        try:
+            guild_id = self._config.getint("Discord", "AllowedGuildID", fallback=0)
+            if guild_id <= 0:
+                errors.append("AllowedGuildID must be a positive integer")
+        except ValueError:
+            errors.append("AllowedGuildID must be a valid integer")
+
+        # Validate API keys have correct prefixes (Pterodactyl and Pelican Panel)
+        client_key = self._config.get("Pterodactyl", "ClientAPIKey", fallback="")
+        if client_key and not (client_key.startswith("ptlc_") or client_key.startswith("pacc_")):
+            errors.append("ClientAPIKey should start with 'ptlc_' (Pterodactyl) or 'pacc_' (Pelican Panel)")
+
+        app_key = self._config.get("Pterodactyl", "ApplicationAPIKey", fallback="")
+        if app_key and not (app_key.startswith("ptla_") or app_key.startswith("papp_")):
+            errors.append("ApplicationAPIKey should start with 'ptla_' (Pterodactyl) or 'papp_' (Pelican Panel)")
+
+        # Validate PanelURL is a valid URL
+        panel_url = self._config.get("Pterodactyl", "PanelURL", fallback="")
+        if panel_url and not (panel_url.startswith("http://") or panel_url.startswith("https://")):
+            errors.append("PanelURL must start with http:// or https://")
+
+        if errors:
+            error_msg = "Configuration validation failed:\n- " + "\n- ".join(errors)
+            logger.error(error_msg)
+            raise ConfigValidationError(error_msg)
+
+        logger.info("Configuration validation passed")
+
+    @property
+    def is_pelican_panel(self) -> bool:
+        """Check if using Pelican Panel API keys (pacc_/papp_ prefixes)."""
+        return (
+            self.client_api_key.startswith("pacc_")
+            or self.application_api_key.startswith("papp_")
+        )
+
+    @property
+    def panel_type(self) -> str:
+        """Return the detected panel type."""
+        return "Pelican Panel" if self.is_pelican_panel else "Pterodactyl"
+
+    def get_panel_display_name(self) -> str:
+        """Get display name for the panel type."""
+        return "Pelican Panel" if self.is_pelican_panel else "Pterodactyl"
+
+    @classmethod
+    def get_instance(cls) -> Optional["Config"]:
+        """Get the singleton config instance."""
+        return cls._instance
+
+
+# Singleton instance for access from views
+Config._instance = None
+
+
+def load_config(config_path: str = "config.ini") -> Config:
+    """
+    Load and validate configuration from file.
+
+    Args:
+        config_path: Path to config.ini file
+
+    Returns:
+        Config: Validated configuration instance
+
+    Raises:
+        ConfigValidationError: If configuration is invalid
+    """
+    config = Config(config_path)
+    Config._instance = config
+    return config
+
+
+def generate_config_from_env(output_path: str = "config.ini") -> None:
+    """
+    Generate config.ini from environment variables.
+
+    Useful for Docker/container deployments where config comes from env vars.
+
+    Required environment variables:
+        DISCORD_TOKEN, ALLOWED_GUILD_ID, PANEL_URL,
+        CLIENT_API_KEY, APPLICATION_API_KEY
+
+    Args:
+        output_path: Path to write generated config.ini
+    """
+    config = configparser.ConfigParser()
+
+    config["Pterodactyl"] = {
+        "PanelURL": os.environ.get("PANEL_URL", ""),
+        "ClientAPIKey": os.environ.get("CLIENT_API_KEY", ""),
+        "ApplicationAPIKey": os.environ.get("APPLICATION_API_KEY", ""),
+    }
+
+    config["Discord"] = {
+        "Token": os.environ.get("DISCORD_TOKEN", ""),
+        "AllowedGuildID": os.environ.get("ALLOWED_GUILD_ID", ""),
+    }
+
+    with open(output_path, "w") as f:
+        config.write(f)
+
+    logger.info(f"Generated config.ini from environment variables at {output_path}")

embeds.py

@@ -0,0 +1,424 @@
+"""
+Embed Management Module
+
+Handles creation, tracking, and persistence of Discord server status embeds.
+Manages embed locations storage and provides utilities for embed lifecycle.
+"""
+
+import json
+import logging
+from datetime import datetime
+from pathlib import Path
+from typing import Any, Dict, List, Optional, Tuple
+
+import discord
+
+from api import PanelAPI
+from config import Config
+from exceptions import EmbedNotFoundError
+from metrics import ServerMetricsManager
+from models import ServerData, ServerResources
+
+logger = logging.getLogger("pterodisbot")
+
+
+class EmbedManager:
+    """
+    Manages server status embeds lifecycle.
+
+    Handles:
+    - Embed creation with real-time server data
+    - Persistent storage of embed locations (channel_id, message_id)
+    - Embed refresh and recreation
+    - Cleanup of deleted/missing embeds
+    - Rate limit protection during bulk operations
+
+    Attributes:
+        config: Application configuration
+        api: Panel API client
+        metrics_manager: Metrics manager for graphs
+        storage_path: Path to embed locations JSON file
+    """
+
+    def __init__(
+        self,
+        config: Config,
+        api: PanelAPI,
+        metrics_manager: ServerMetricsManager,
+    ) -> None:
+        """
+        Initialize the embed manager.
+
+        Args:
+            config: Validated application configuration
+            api: Initialized PanelAPI instance
+            metrics_manager: ServerMetricsManager instance
+        """
+        self.config = config
+        self.api = api
+        self.metrics_manager = metrics_manager
+        self.storage_path = config.embed_storage_path
+        self.embed_locations: Dict[str, Dict[str, int]] = {}
+        self.previous_states: Dict[str, Tuple[str, float, Optional[float]]] = {}
+
+        # Ensure storage directory exists
+        self.storage_path.parent.mkdir(parents=True, exist_ok=True)
+        logger.info("Initialized EmbedManager")
+
+    async def load_locations(self) -> None:
+        """Load saved embed locations from JSON storage file."""
+        logger.debug("Loading embed locations from storage")
+        if not self.storage_path.exists():
+            logger.info("No existing embed locations file found")
+            return
+
+        try:
+            with open(self.storage_path, "r") as f:
+                self.embed_locations = json.load(f)
+            logger.info(f"Loaded {len(self.embed_locations)} embed locations from storage")
+        except Exception as e:
+            logger.error(f"Failed to load embed locations: {str(e)}")
+
+    async def save_locations(self) -> None:
+        """Save current embed locations to JSON storage file."""
+        logger.debug("Saving embed locations to storage")
+        try:
+            with open(self.storage_path, "w") as f:
+                json.dump(self.embed_locations, f, indent=2)
+            logger.debug("Successfully saved embed locations")
+        except Exception as e:
+            logger.error(f"Failed to save embed locations: {str(e)}")
+
+    async def track_embed(self, server_id: str, message: discord.Message) -> None:
+        """
+        Track a newly created embed in storage.
+
+        Args:
+            server_id: The server's panel identifier
+            message: Discord message containing the embed
+        """
+        logger.debug(f"Tracking new embed for server {server_id} in channel {message.channel.id}")
+        self.embed_locations[server_id] = {
+            "channel_id": str(message.channel.id),
+            "message_id": str(message.id),
+        }
+        await self.save_locations()
+
+    async def untrack_embed(self, server_id: str) -> None:
+        """
+        Remove a server's embed from tracking.
+
+        Args:
+            server_id: The server's panel identifier
+        """
+        self.embed_locations.pop(server_id, None)
+        self.previous_states.pop(server_id, None)
+        await self.save_locations()
+
+    async def create_server_embed(
+        self, server_data: ServerData, resources: ServerResources
+    ) -> Tuple[discord.Embed, discord.ui.View]:
+        """
+        Create a status embed and interactive view for a server.
+
+        Args:
+            server_data: Server information from panel
+            resources: Current resource usage data
+
+        Returns:
+            Tuple of (embed, view) ready for display
+        """
+        attributes = server_data.attributes
+        identifier = attributes.identifier
+        name = attributes.name
+        description = attributes.description or "No description available"
+
+        logger.debug(f"Building status embed for server {name} ({identifier})")
+
+        # Parse resource data
+        current_state = resources.attributes.current_state.title()
+        is_suspended = attributes.suspended
+
+        # Create embed with appropriate color based on status
+        embed = discord.Embed(
+            title=f"{name} - {current_state}",
+            description=description,
+            color=(
+                discord.Color.blue()
+                if current_state.lower() == "running"
+                else discord.Color.red()
+            ),
+            timestamp=datetime.now(),
+        )
+
+        embed.add_field(name="🆔 Server ID", value=f"`{identifier}`", inline=True)
+
+        if is_suspended:
+            embed.add_field(name="ℹ️ Status", value="⛔ `Suspended`", inline=True)
+        else:
+            embed.add_field(name="ℹ️ Status", value="✅ `Active`", inline=True)
+
+        # Add resource usage if server is running
+        if current_state.lower() != "offline":
+            embed = self._add_resource_fields(embed, resources, attributes)
+            embed = self._add_graph_fields(embed, identifier)
+
+        embed.set_footer(text="Last updated")
+
+        # Create interactive view with control buttons
+        from views import ServerStatusView
+
+        view = ServerStatusView(
+            server_id=identifier,
+            server_name=name,
+            api=self.api,
+            server_data=server_data.model_dump(),
+        )
+
+        logger.debug(f"Successfully built status components for {name}")
+        return embed, view
+
+    def _add_resource_fields(
+        self, embed: discord.Embed, resources: ServerResources, attributes: Any
+    ) -> discord.Embed:
+        """Add resource usage fields to embed."""
+        resource_attrs = resources.attributes.resources
+        limits = attributes.limits
+
+        # Current usage
+        cpu_usage = round(resource_attrs.cpu_absolute, 2)
+        memory_usage = round(resource_attrs.memory_bytes / (1024**2), 2)
+        disk_usage = round(resource_attrs.disk_bytes / (1024**2), 2)
+        network_rx = round(resource_attrs.network_rx_bytes / (1024**2), 2)
+        network_tx = round(resource_attrs.network_tx_bytes / (1024**2), 2)
+
+        # Maximum allocated resources
+        cpu_limit = limits.cpu
+        memory_limit = limits.memory
+        disk_limit = limits.disk
+
+        # Format limit values - display ∞ for unlimited (0 limit)
+        def format_limit(value: int, unit: str = "") -> str:
+            if value == 0:
+                return f"{'∞':>8} {unit}"
+            return f"{value:>8} {unit}"
+
+        # Format uptime
+        uptime_ms = resource_attrs.uptime
+        if uptime_ms > 0:
+            uptime_seconds = uptime_ms // 1000
+            if uptime_seconds < 60:
+                uptime_text = f"`{uptime_seconds}s`"
+            elif uptime_seconds < 3600:
+                uptime_text = f"`{uptime_seconds // 60}m {uptime_seconds % 60}s`"
+            elif uptime_seconds < 86400:
+                hours = uptime_seconds // 3600
+                minutes = (uptime_seconds % 3600) // 60
+                uptime_text = f"`{hours}h {minutes}m`"
+            else:
+                days = uptime_seconds // 86400
+                hours = (uptime_seconds % 86400) // 3600
+                uptime_text = f"`{days}d {hours}h`"
+        else:
+            uptime_text = "`Just started`"
+
+        embed.add_field(name="⏱️ Uptime", value=uptime_text, inline=True)
+
+        # Resource usage text box
+        usage_text = (
+            f"```properties\n"
+            f"CPU    : [{cpu_usage:>8} / {format_limit(cpu_limit, '%')}\n"
+            f"Memory : [{memory_usage:>8} / {format_limit(memory_limit, 'MiB')}\n"
+            f"Disk   : [{disk_usage:>8} / {format_limit(disk_limit, 'MiB')}\n"
+            f"```"
+        )
+
+        embed.add_field(name="📊 Resource Usage", value=usage_text, inline=False)
+        embed.add_field(name="Network In", value=f"📥 `{network_rx} MiB`", inline=True)
+        embed.add_field(name="Network Out", value=f"📤 `{network_tx} MiB`", inline=True)
+
+        return embed
+
+    def _add_graph_fields(
+        self, embed: discord.Embed, server_id: str
+    ) -> discord.Embed:
+        """Add metrics graph fields to embed if data available."""
+        server_graphs = self.metrics_manager.get_server_graphs(server_id)
+        if server_graphs and server_graphs.has_sufficient_data:
+            summary = server_graphs.get_data_summary()
+
+            graph_description = (
+                f">>> `Data points: {summary['point_count']}/6`\n"
+                f"`CPU trend: {summary['cpu_trend']} • Memory trend: {summary['memory_trend']}`"
+            )
+
+            embed.add_field(
+                name="📈 Usage Trends (Last Minute)",
+                value=graph_description,
+                inline=False,
+            )
+            embed.set_image(url=f"attachment://metrics_graph_{server_id}.png")
+
+        return embed
+
+    async def refresh_all_embeds(self) -> Tuple[int, int]:
+        """
+        Perform a complete refresh of all server status embeds.
+
+        Creates new embeds and deletes old ones to prevent duplication.
+
+        Returns:
+            Tuple of (deleted_count, created_count)
+        """
+        logger.info("Starting full refresh of all server embeds")
+        from bot import PterodactylBot  # Avoid circular import
+        bot = PterodactylBot.get_instance()
+
+        deleted_count = 0
+        created_count = 0
+        skipped_count = 0
+
+        try:
+            # Get current server list
+            servers = await self.api.get_servers()
+            if not servers:
+                logger.warning("No servers found in panel during refresh")
+                return 0, 0
+
+            server_cache = {s["attributes"]["identifier"]: s for s in servers}
+
+            # Create new embeds
+            new_embeds = {}
+
+            for server_id, server_data in server_cache.items():
+                if server_id not in self.embed_locations:
+                    skipped_count += 1
+                    continue
+
+                channel_id = self.embed_locations[server_id]["channel_id"]
+                channel = bot.get_channel(int(channel_id))
+                if not channel:
+                    logger.warning(f"Channel {channel_id} not found for server {server_id}")
+                    continue
+
+                try:
+                    resources = await self.api.get_server_resources(server_id)
+                    server_obj = ServerData.model_validate(server_data)
+                    resources_obj = ServerResources.model_validate(resources)
+
+                    embed, view = await self.create_server_embed(server_obj, resources_obj)
+                    message = await channel.send(embed=embed, view=view)
+
+                    new_embeds[server_id] = {
+                        "channel_id": str(channel.id),
+                        "message_id": str(message.id),
+                    }
+                    created_count += 1
+                    logger.info(f"Created new embed for server {server_data['attributes']['name']}")
+
+                    await asyncio.sleep(1)  # Rate limit protection
+                except Exception as e:
+                    logger.error(f"Failed to create new embed for server {server_id}: {str(e)}")
+
+            if not new_embeds:
+                logger.warning("No new embeds created during refresh")
+                return 0, 0
+
+            # Delete old embeds
+            for server_id, location in list(self.embed_locations.items()):
+                try:
+                    channel = bot.get_channel(int(location["channel_id"]))
+                    if channel:
+                        try:
+                            message = await channel.fetch_message(int(location["message_id"]))
+                            await message.delete()
+                            deleted_count += 1
+                            logger.debug(f"Deleted old embed for server {server_id}")
+                            await asyncio.sleep(0.5)
+                        except discord.NotFound:
+                            logger.debug(f"Old embed for server {server_id} already deleted")
+                        except Exception as e:
+                            logger.error(f"Failed to delete old embed for server {server_id}: {str(e)}")
+                except Exception as e:
+                    logger.error(f"Error processing old embed for server {server_id}: {str(e)}")
+
+            logger.info(f"Deleted {deleted_count} old embeds, created {created_count} new embeds")
+
+            # Update storage
+            self.embed_locations = new_embeds
+            await self.save_locations()
+
+            return deleted_count, created_count
+
+        except Exception as e:
+            logger.error(f"Critical error during embed refresh: {str(e)}")
+            raise
+
+    async def update_single_embed(
+        self,
+        server_id: str,
+        location: Dict[str, str],
+        server_data: ServerData,
+        resources: ServerResources,
+        current_state: str,
+        cpu_usage: float,
+    ) -> bool:
+        """
+        Update a single server's status embed.
+
+        Args:
+            server_id: Server identifier
+            location: Embed location dict with channel_id and message_id
+            server_data: Current server data
+            resources: Current resource data
+            current_state: Current power state
+            cpu_usage: Current CPU usage
+
+        Returns:
+            True if embed was updated, False otherwise
+        """
+        from bot import PterodactylBot
+        bot = PterodactylBot.get_instance()
+
+        try:
+            embed, view = await self.create_server_embed(server_data, resources)
+
+            channel = bot.get_channel(int(location["channel_id"]))
+            if not channel:
+                logger.warning(f"Channel {location['channel_id']} not found for server {server_id}")
+                return False
+
+            message = await channel.fetch_message(int(location["message_id"]))
+
+            # Handle graph attachments
+            files = []
+            server_graphs = self.metrics_manager.get_server_graphs(server_id)
+
+            if current_state == "running" and server_graphs and server_graphs.has_sufficient_data:
+                combined_graph = server_graphs.generate_combined_graph()
+                if combined_graph:
+                    files.append(
+                        discord.File(
+                            combined_graph,
+                            filename=f"metrics_graph_{server_id}.png",
+                        )
+                    )
+            # Empty files list removes existing attachments
+
+            await message.edit(embed=embed, view=view, attachments=files)
+            logger.debug(f"Updated status for {server_data.attributes.name}")
+            return True
+
+        except discord.NotFound:
+            logger.warning(f"Embed for server {server_id} not found, removing from tracking")
+            self.embed_locations.pop(server_id, None)
+            self.previous_states.pop(server_id, None)
+            await self.save_locations()
+            return False
+        except Exception as e:
+            logger.error(f"Failed to update status for server {server_id}: {str(e)}")
+            return False
+
+
+# Import at bottom to avoid circular imports
+import asyncio

entrypoint.sh

@@ -5,4 +5,4 @@ if ! python generate_config.py; then
     exit 1
 fi
 
-exec python pterodisbot.py
+exec python main.py

exceptions.py

@@ -0,0 +1,84 @@
+"""
+Custom Exception Classes
+
+Defines application-specific exceptions for better error handling
+and clearer error messages throughout the codebase.
+"""
+
+
+class PterodisBotError(Exception):
+    """Base exception class for all PterodisBot errors."""
+
+    pass
+
+
+class ConfigValidationError(PterodisBotError):
+    """
+    Raised when configuration validation fails.
+
+    Contains detailed error messages for all validation failures.
+    """
+
+    pass
+
+
+class APIError(PterodisBotError):
+    """
+    Raised when an API request fails.
+
+    Attributes:
+        endpoint: The API endpoint that failed
+        status_code: HTTP status code if available
+        message: Error message from the API
+    """
+
+    def __init__(
+        self,
+        message: str,
+        endpoint: str = "",
+        status_code: int = 0,
+    ) -> None:
+        super().__init__(message)
+        self.endpoint = endpoint
+        self.status_code = status_code
+        self.message = message
+
+
+class ServerNotFoundError(PterodisBotError):
+    """
+    Raised when a requested server is not found in the panel.
+    """
+
+    pass
+
+
+class PermissionError(PterodisBotError):
+    """
+    Raised when a user lacks required permissions for an action.
+    """
+
+    pass
+
+
+class EmbedNotFoundError(PterodisBotError):
+    """
+    Raised when a tracked embed message is not found in Discord.
+    """
+
+    pass
+
+
+class MetricsError(PterodisBotError):
+    """
+    Raised when metrics graph generation fails.
+    """
+
+    pass
+
+
+class ConfigurationError(PterodisBotError):
+    """
+    Raised when there's a configuration issue not caught during startup validation.
+    """
+
+    pass

generate_config.py

@@ -3,6 +3,7 @@ import configparser
 from pathlib import Path
 from dotenv import load_dotenv
 
+
 def validate_env_vars():
     """Validate that required environment variables are set and non-empty."""
     required_vars = [
@@ -17,33 +18,82 @@ def validate_env_vars():
     if missing_vars:
         raise ValueError(f"Missing required environment variables: {', '.join(missing_vars)}")
 
+
+def get_env_config():
+    """Extract configuration from environment variables."""
+    return {
+        'Discord': {
+            'Token': os.getenv('DISCORD_TOKEN'),
+            'AllowedGuildID': os.getenv('ALLOWED_GUILD_ID')
+        },
+        'Pterodactyl': {
+            'PanelURL': os.getenv('PANEL_URL'),
+            'ClientAPIKey': os.getenv('CLIENT_API_KEY'),
+            'ApplicationAPIKey': os.getenv('APPLICATION_API_KEY')
+        }
+    }
+
+
+def configs_match(existing_config, env_config):
+    """Compare existing config with environment variables."""
+    for section, values in env_config.items():
+        if not existing_config.has_section(section):
+            return False
+        for key, env_value in values.items():
+            if env_value is None:
+                # Environment variable not set, don't require match
+                continue
+            existing_value = existing_config.get(section, key, fallback=None)
+            if existing_value != env_value:
+                return False
+    return True
+
+
 def generate_config():
-    """Generate config.ini from environment variables."""
-    # Validate environment variables first
-    validate_env_vars()
+    """
+    Generate config.ini from environment variables if:
+    1. No config file exists, OR
+    2. Config file exists but values don't match environment variables
     
-    config = configparser.ConfigParser()
-    
-    # Discord section - ensure all values are strings
-    config['Discord'] = {
-        'Token': str(os.getenv('DISCORD_TOKEN')),
-        'AllowedGuildID': str(os.getenv('ALLOWED_GUILD_ID'))
-    }
-    
-    # Pterodactyl section - ensure all values are strings
-    config['Pterodactyl'] = {
-        'PanelURL': str(os.getenv('PANEL_URL')),
-        'ClientAPIKey': str(os.getenv('CLIENT_API_KEY')),
-        'ApplicationAPIKey': str(os.getenv('APPLICATION_API_KEY'))
-    }
-    
-    # Write to file
+    Environment variables always take priority.
+    """
     config_path = Path(os.getenv('CONFIG_PATH', 'config.ini'))
+    
+    # Check if config file exists
+    if not config_path.exists():
+        print(f"No config file found at {config_path}, generating from environment variables...")
+        validate_env_vars()
+        _write_config(config_path, get_env_config())
+        return
+    
+    # Config file exists, read it and compare with environment
+    existing_config = configparser.ConfigParser()
+    existing_config.read(config_path)
+    
+    env_config = get_env_config()
+    
+    if not configs_match(existing_config, env_config):
+        print(f"Config file at {config_path} differs from environment variables, regenerating...")
+        # Validate that required env vars are present for regeneration
+        validate_env_vars()
+        _write_config(config_path, env_config)
+        return
+    
+    print(f"Config file at {config_path} matches environment variables, using existing config.")
+
+
+def _write_config(config_path, config_dict):
+    """Write config dictionary to file."""
+    config = configparser.ConfigParser()
+    for section, values in config_dict.items():
+        config[section] = {k: str(v) for k, v in values.items() if v is not None}
+    
     config_path.parent.mkdir(exist_ok=True, parents=True)
     
     with open(config_path, 'w') as configfile:
         config.write(configfile)
-    print(f"Config file generated at {config_path}")
+    print(f"Config file written to {config_path}")
+
 
 if __name__ == "__main__":
     load_dotenv()

logging_config.py

@@ -0,0 +1,66 @@
+"""
+Logging Configuration Module
+
+Sets up application-wide logging with:
+- Rotating file handler (5MB max, 3 backups) at DEBUG level
+- Console handler at INFO level
+- Consistent formatting for both handlers
+"""
+
+import logging
+import os
+from logging.handlers import RotatingFileHandler
+from pathlib import Path
+
+
+def setup_logging(log_dir: str = "logs", log_file: str = "pterodisbot.log") -> logging.Logger:
+    """
+    Configure application-wide logging.
+
+    Args:
+        log_dir: Directory for log files (relative to this file)
+        log_file: Name of the log file
+
+    Returns:
+        Configured logger instance
+    """
+    # Create logs directory if it doesn't exist
+    logs_path = Path(__file__).parent / log_dir
+    logs_path.mkdir(parents=True, exist_ok=True)
+
+    # Get or create the application logger
+    logger = logging.getLogger("pterodisbot")
+    logger.setLevel(logging.DEBUG)
+
+    # Avoid adding duplicate handlers if called multiple times
+    if logger.handlers:
+        return logger
+
+    # File handler for logs (rotates when reaching 5MB, keeps 3 backups)
+    file_handler = RotatingFileHandler(
+        filename=logs_path / log_file,
+        maxBytes=5 * 1024 * 1024,  # 5 MiB max log file size
+        backupCount=3,  # Rotate through 3 files
+        encoding="utf-8",
+    )
+    file_handler.setLevel(logging.DEBUG)
+    file_handler.setFormatter(
+        logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+    )
+    logger.addHandler(file_handler)
+
+    # Console handler for real-time output (INFO level and above)
+    console_handler = logging.StreamHandler()
+    console_handler.setLevel(logging.INFO)
+    console_handler.setFormatter(
+        logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
+    )
+    logger.addHandler(console_handler)
+
+    logger.info("Initialized logging system with file and console output")
+    return logger
+
+
+def get_logger() -> logging.Logger:
+    """Get the configured logger instance."""
+    return logging.getLogger("pterodisbot")

main.py

@@ -0,0 +1,55 @@
+#!/usr/bin/env python3
+"""
+Main entry point for the Pterodactyl Discord Bot.
+
+This module replaces the original pterodisbot.py as the entry point.
+It loads configuration and starts the modular bot application.
+"""
+
+import asyncio
+import logging
+import sys
+
+from bot import run_bot
+from config import load_config, ConfigValidationError
+from exceptions import PterodisBotError
+from logging_config import setup_logging
+
+# Configure application logging
+setup_logging()
+
+logger = logging.getLogger("pterodisbot")
+
+
+def main() -> int:
+    """
+    Synchronous entry point for the bot application.
+
+    Returns:
+        Exit code (0 for success, 1 for error)
+    """
+    try:
+        # Load and validate configuration
+        logger.info("Loading configuration...")
+        config = load_config("config.ini")
+        logger.info(f"Configuration loaded - Panel type: {config.get_panel_display_name()}")
+
+        # Run the bot
+        asyncio.run(run_bot(config))
+        return 0
+
+    except ConfigValidationError as e:
+        logger.critical(f"Configuration error: {str(e)}")
+        return 1
+    except PterodisBotError as e:
+        logger.critical(f"Application error: {str(e)}")
+        return 1
+    except KeyboardInterrupt:
+        pass  # Already logged by signal handler
+    except Exception as e:
+        logger.critical(f"Unexpected error during startup: {str(e)}", exc_info=True)
+        return 1
+
+
+if __name__ == "__main__":
+    sys.exit(main())

metrics.py

@@ -0,0 +1,538 @@
+"""
+Server Metrics Graphs Module
+
+Provides graphing capabilities for server CPU and memory usage.
+Generates line graphs as PNG images for embedding in Discord messages.
+"""
+
+import io
+import logging
+import math
+from collections import deque
+from datetime import datetime
+from typing import Any, Dict, Optional
+
+import matplotlib
+import matplotlib.dates as mdates
+import matplotlib.pyplot as plt
+
+matplotlib.use("Agg")  # Use non-interactive backend for server environments
+
+logger = logging.getLogger("pterodisbot")
+
+
+class ServerMetricsGraphs:
+    """
+    Manages CPU and memory usage graphs for individual servers.
+
+    Features:
+    - Stores last 6 data points (1 minute of history at 10-second intervals)
+    - Generates PNG images of line graphs for Discord embedding
+    - Automatic data rotation (FIFO queue with max 6 points)
+    - Separate tracking for CPU percentage and memory MB usage
+    - Dynamic CPU scaling in 100% increments for multi-vCPU servers
+    - Clean graph styling optimized for Discord dark theme
+    """
+
+    def __init__(self, server_id: str, server_name: str) -> None:
+        """
+        Initialize metrics tracking for a server.
+
+        Args:
+            server_id: Panel server identifier
+            server_name: Human-readable server name
+        """
+        self.server_id = server_id
+        self.server_name = server_name
+
+        # Use deque with maxlen=6 for automatic FIFO rotation
+        # Each entry is a tuple: (timestamp, cpu_percent, memory_mb)
+        self.data_points = deque(maxlen=6)
+
+        # Track if we have enough data for meaningful graphs (at least 2 points)
+        self.has_sufficient_data = False
+
+        logger.debug(f"Initialized metrics tracking for server {server_name} ({server_id})")
+
+    def add_data_point(
+        self, cpu_percent: float, memory_mb: float, timestamp: Optional[datetime] = None
+    ) -> None:
+        """
+        Add a new data point to the metrics history.
+
+        Args:
+            cpu_percent: Current CPU usage percentage
+            memory_mb: Current memory usage in megabytes
+            timestamp: Optional timestamp, defaults to current time
+        """
+        if timestamp is None:
+            timestamp = datetime.now()
+
+        # Add new data point (automatically rotates old data due to maxlen=6)
+        self.data_points.append((timestamp, cpu_percent, memory_mb))
+
+        # Update sufficient data flag
+        self.has_sufficient_data = len(self.data_points) >= 2
+
+        logger.debug(
+            f"Added metrics data point for {self.server_name}: "
+            f"CPU={cpu_percent}%, Memory={memory_mb}MB"
+        )
+
+    def _calculate_cpu_scale_limit(self, max_cpu_value: float) -> int:
+        """
+        Calculate appropriate CPU scale limit in 100% increments.
+
+        Args:
+            max_cpu_value: Maximum CPU value in the dataset
+
+        Returns:
+            Scale limit rounded up to nearest 100% increment
+        """
+        if max_cpu_value <= 100:
+            return 100
+
+        # Round up to nearest 100% increment
+        # e.g., 150% -> 200%, 250% -> 300%, 350% -> 400%
+        return math.ceil(max_cpu_value / 100) * 100
+
+    def generate_cpu_graph(self) -> Optional[io.BytesIO]:
+        """
+        Generate a CPU usage line graph as a PNG image.
+
+        Returns:
+            BytesIO object containing PNG image data, or None if insufficient data
+        """
+        if not self.has_sufficient_data:
+            logger.debug(f"Insufficient data for CPU graph generation: {self.server_name}")
+            return None
+
+        try:
+            # Extract timestamps and CPU data
+            timestamps = [point[0] for point in self.data_points]
+            cpu_values = [point[1] for point in self.data_points]
+
+            # Calculate dynamic CPU scale limit
+            max_cpu = max(cpu_values)
+            cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
+
+            # Create figure with dark theme styling
+            plt.style.use("dark_background")
+            fig, ax = plt.subplots(figsize=(8, 4), dpi=100)
+            fig.patch.set_facecolor("#2f3136")  # Discord dark theme background
+            ax.set_facecolor("#36393f")  # Slightly lighter for graph area
+
+            # Plot CPU line with gradient fill
+            ax.fill_between(timestamps, cpu_values, alpha=0.3, color="#7289da")
+
+            # Customize axes with dynamic scaling
+            ax.set_ylabel("CPU Usage (%)", color="#ffffff", fontsize=10)
+            ax.set_ylim(0, cpu_scale_limit)
+
+            # Add horizontal grid lines at 100% increments for better readability
+            for i in range(100, cpu_scale_limit + 1, 100):
+                ax.axhline(y=i, color="#ffffff", alpha=0.2, linestyle="--", linewidth=0.8)
+
+            # Format time axis
+            ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
+            ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
+            plt.setp(
+                ax.xaxis.get_majorticklabels(),
+                rotation=45,
+                ha="right",
+                color="#ffffff",
+                fontsize=8,
+            )
+
+            # Style the graph
+            ax.tick_params(colors="#ffffff", labelsize=8)
+            ax.grid(True, alpha=0.3, color="#ffffff")
+            ax.spines["bottom"].set_color("#ffffff")
+            ax.spines["left"].set_color("#ffffff")
+            ax.spines["top"].set_visible(False)
+            ax.spines["right"].set_visible(False)
+
+            # Add title with scale info for multi-vCPU servers
+            title = f"{self.server_name} - CPU Usage"
+            if cpu_scale_limit > 100:
+                estimated_vcpus = cpu_scale_limit // 100
+                title += f" (~{estimated_vcpus} vCPU cores)"
+            ax.set_title(title, color="#ffffff", fontsize=12, pad=20)
+
+            # Tight layout to prevent label cutoff
+            plt.tight_layout()
+
+            # Save to BytesIO
+            img_buffer = io.BytesIO()
+            plt.savefig(
+                img_buffer,
+                format="png",
+                facecolor="#2f3136",
+                edgecolor="none",
+                bbox_inches="tight",
+                dpi=100,
+            )
+            img_buffer.seek(0)
+
+            # Clean up matplotlib resources
+            plt.close(fig)
+
+            logger.debug(f"Generated CPU graph for {self.server_name} (scale: 0-{cpu_scale_limit}%)")
+            return img_buffer
+
+        except Exception as e:
+            logger.error(f"Failed to generate CPU graph for {self.server_name}: {str(e)}")
+            plt.close("all")  # Clean up any remaining figures
+            return None
+
+    def generate_memory_graph(self) -> Optional[io.BytesIO]:
+        """
+        Generate a memory usage line graph as a PNG image.
+
+        Returns:
+            BytesIO object containing PNG image data, or None if insufficient data
+        """
+        if not self.has_sufficient_data:
+            logger.debug(f"Insufficient data for memory graph generation: {self.server_name}")
+            return None
+
+        try:
+            # Extract timestamps and memory data
+            timestamps = [point[0] for point in self.data_points]
+            memory_values = [point[2] for point in self.data_points]
+
+            # Create figure with dark theme styling
+            plt.style.use("dark_background")
+            fig, ax = plt.subplots(figsize=(8, 4), dpi=100)
+            fig.patch.set_facecolor("#2f3136")  # Discord dark theme background
+            ax.set_facecolor("#36393f")  # Slightly lighter for graph area
+
+            # Plot memory line with gradient fill
+            ax.fill_between(timestamps, memory_values, alpha=0.3, color="#43b581")
+
+            # Customize axes
+            ax.set_ylabel("Memory Usage (MB)", color="#ffffff", fontsize=10)
+            ax.set_ylim(0, max(memory_values) * 1.1)  # Dynamic scaling with 10% padding
+
+            # Format time axis
+            ax.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
+            ax.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
+            plt.setp(
+                ax.xaxis.get_majorticklabels(),
+                rotation=45,
+                ha="right",
+                color="#ffffff",
+                fontsize=8,
+            )
+
+            # Style the graph
+            ax.tick_params(colors="#ffffff", labelsize=8)
+            ax.grid(True, alpha=0.3, color="#ffffff")
+            ax.spines["bottom"].set_color("#ffffff")
+            ax.spines["left"].set_color("#ffffff")
+            ax.spines["top"].set_visible(False)
+            ax.spines["right"].set_visible(False)
+
+            # Add title
+            ax.set_title(
+                f"{self.server_name} - Memory Usage",
+                color="#ffffff",
+                fontsize=12,
+                pad=20,
+            )
+
+            # Tight layout to prevent label cutoff
+            plt.tight_layout()
+
+            # Save to BytesIO
+            img_buffer = io.BytesIO()
+            plt.savefig(
+                img_buffer,
+                format="png",
+                facecolor="#2f3136",
+                edgecolor="none",
+                bbox_inches="tight",
+                dpi=100,
+            )
+            img_buffer.seek(0)
+
+            # Clean up matplotlib resources
+            plt.close(fig)
+
+            logger.debug(f"Generated memory graph for {self.server_name}")
+            return img_buffer
+
+        except Exception as e:
+            logger.error(f"Failed to generate memory graph for {self.server_name}: {str(e)}")
+            plt.close("all")  # Clean up any remaining figures
+            return None
+
+    def generate_combined_graph(self) -> Optional[io.BytesIO]:
+        """
+        Generate a combined CPU and memory usage graph as a PNG image.
+
+        Returns:
+            BytesIO object containing PNG image data, or None if insufficient data
+        """
+        if not self.has_sufficient_data:
+            logger.debug(f"Insufficient data for combined graph generation: {self.server_name}")
+            return None
+
+        try:
+            # Extract data
+            timestamps = [point[0] for point in self.data_points]
+            cpu_values = [point[1] for point in self.data_points]
+            memory_values = [point[2] for point in self.data_points]
+
+            # Calculate dynamic CPU scale limit
+            max_cpu = max(cpu_values)
+            cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
+
+            # Create figure with two subplots
+            plt.style.use("dark_background")
+            fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(8, 6), dpi=100, sharex=True)
+            fig.patch.set_facecolor("#2f3136")
+
+            # CPU subplot
+            ax1.set_facecolor("#36393f")
+            ax1.plot(
+                timestamps,
+                cpu_values,
+                color="#7289da",
+                linewidth=2.5,
+                marker="o",
+                markersize=4,
+            )
+            ax1.fill_between(timestamps, cpu_values, alpha=0.3, color="#7289da")
+            ax1.set_ylabel("CPU Usage (%)", color="#ffffff", fontsize=10)
+            ax1.set_ylim(0, cpu_scale_limit)
+            ax1.tick_params(colors="#ffffff", labelsize=8)
+            ax1.grid(True, alpha=0.3, color="#ffffff")
+
+            # Add horizontal grid lines at 100% increments for CPU subplot
+            for i in range(100, cpu_scale_limit + 1, 100):
+                ax1.axhline(y=i, color="#ffffff", alpha=0.2, linestyle="--", linewidth=0.8)
+
+            # Title with vCPU info if applicable
+            title = f"{self.server_name} - Resource Usage"
+            if cpu_scale_limit > 100:
+                estimated_vcpus = cpu_scale_limit // 100
+                title += f" (~{estimated_vcpus} vCPU cores)"
+            ax1.set_title(title, color="#ffffff", fontsize=12)
+
+            # Memory subplot
+            ax2.set_facecolor("#36393f")
+            ax2.plot(
+                timestamps,
+                memory_values,
+                color="#43b581",
+                linewidth=2.5,
+                marker="o",
+                markersize=4,
+            )
+            ax2.fill_between(timestamps, memory_values, alpha=0.3, color="#43b581")
+            ax2.set_ylabel("Memory (MB)", color="#ffffff", fontsize=10)
+            ax2.set_ylim(0, max(memory_values) * 1.1)
+            ax2.tick_params(colors="#ffffff", labelsize=8)
+            ax2.grid(True, alpha=0.3, color="#ffffff")
+
+            # Format time axis (only on bottom subplot)
+            ax2.xaxis.set_major_formatter(mdates.DateFormatter("%H:%M:%S"))
+            ax2.xaxis.set_major_locator(mdates.SecondLocator(interval=20))
+            plt.setp(
+                ax2.xaxis.get_majorticklabels(),
+                rotation=45,
+                ha="right",
+                color="#ffffff",
+                fontsize=8,
+            )
+
+            # Style both subplots
+            for ax in [ax1, ax2]:
+                ax.spines["bottom"].set_color("#ffffff")
+                ax.spines["left"].set_color("#ffffff")
+                ax.spines["top"].set_visible(False)
+                ax.spines["right"].set_visible(False)
+
+            plt.tight_layout()
+
+            # Save to BytesIO
+            img_buffer = io.BytesIO()
+            plt.savefig(
+                img_buffer,
+                format="png",
+                facecolor="#2f3136",
+                edgecolor="none",
+                bbox_inches="tight",
+                dpi=100,
+            )
+            img_buffer.seek(0)
+
+            plt.close(fig)
+
+            logger.debug(f"Generated combined graph for {self.server_name} (CPU scale: 0-{cpu_scale_limit}%)")
+            return img_buffer
+
+        except Exception as e:
+            logger.error(f"Failed to generate combined graph for {self.server_name}: {str(e)}")
+            plt.close("all")
+            return None
+
+    def get_data_summary(self) -> Dict[str, Any]:
+        """
+        Get summary statistics for the current data points.
+
+        Returns:
+            Dictionary containing data point count, latest values, and trends
+        """
+        if not self.data_points:
+            return {
+                "point_count": 0,
+                "has_data": False,
+                "latest_cpu": 0,
+                "latest_memory": 0,
+            }
+
+        # Get latest values
+        latest_point = self.data_points[-1]
+        latest_cpu = latest_point[1]
+        latest_memory = latest_point[2]
+
+        # Calculate CPU scale info
+        max_cpu = max(point[1] for point in self.data_points)
+        cpu_scale_limit = self._calculate_cpu_scale_limit(max_cpu)
+        estimated_vcpus = cpu_scale_limit // 100
+
+        # Calculate trends if we have multiple points
+        cpu_trend = "stable"
+        memory_trend = "stable"
+
+        if len(self.data_points) >= 2:
+            first_point = self.data_points[0]
+            cpu_change = latest_cpu - first_point[1]
+            memory_change = latest_memory - first_point[2]
+
+            # Determine trends (>5% change considered significant)
+            if abs(cpu_change) > 5:
+                cpu_trend = "increasing" if cpu_change > 0 else "decreasing"
+
+            if abs(memory_change) > 50:  # 50MB change threshold
+                memory_trend = "increasing" if memory_change > 0 else "decreasing"
+
+        return {
+            "point_count": len(self.data_points),
+            "has_data": self.has_sufficient_data,
+            "latest_cpu": latest_cpu,
+            "latest_memory": latest_memory,
+            "cpu_trend": cpu_trend,
+            "memory_trend": memory_trend,
+            "cpu_scale_limit": cpu_scale_limit,
+            "estimated_vcpus": estimated_vcpus,
+            "time_span_minutes": len(self.data_points) * 10 / 60,  # Convert to minutes
+        }
+
+
+class ServerMetricsManager:
+    """
+    Global manager for all server metrics graphs.
+
+    Handles:
+    - Creation and cleanup of ServerMetricsGraphs instances
+    - Bulk operations across all tracked servers
+    - Memory management for graph storage
+    """
+
+    def __init__(self) -> None:
+        """Initialize the metrics manager."""
+        self.server_graphs: Dict[str, ServerMetricsGraphs] = {}
+        logger.info("Initialized ServerMetricsManager")
+
+    def get_or_create_server_graphs(self, server_id: str, server_name: str) -> ServerMetricsGraphs:
+        """
+        Get existing ServerMetricsGraphs instance or create a new one.
+
+        Args:
+            server_id: Panel server identifier
+            server_name: Human-readable server name
+
+        Returns:
+            ServerMetricsGraphs instance for the specified server
+        """
+        if server_id not in self.server_graphs:
+            self.server_graphs[server_id] = ServerMetricsGraphs(server_id, server_name)
+            logger.debug(f"Created new metrics graphs for server {server_name}")
+
+        return self.server_graphs[server_id]
+
+    def add_server_data(
+        self, server_id: str, server_name: str, cpu_percent: float, memory_mb: float
+    ) -> None:
+        """
+        Add data point to a server's metrics tracking.
+
+        Args:
+            server_id: Panel server identifier
+            server_name: Human-readable server name
+            cpu_percent: Current CPU usage percentage
+            memory_mb: Current memory usage in megabytes
+        """
+        graphs = self.get_or_create_server_graphs(server_id, server_name)
+        graphs.add_data_point(cpu_percent, memory_mb)
+
+    def remove_server(self, server_id: str) -> None:
+        """
+        Remove a server from metrics tracking.
+
+        Args:
+            server_id: Panel server identifier to remove
+        """
+        if server_id in self.server_graphs:
+            del self.server_graphs[server_id]
+            logger.debug(f"Removed metrics tracking for server {server_id}")
+
+    def get_server_graphs(self, server_id: str) -> Optional[ServerMetricsGraphs]:
+        """
+        Get ServerMetricsGraphs instance for a specific server.
+
+        Args:
+            server_id: Panel server identifier
+
+        Returns:
+            ServerMetricsGraphs instance or None if not found
+        """
+        return self.server_graphs.get(server_id)
+
+    def cleanup_old_servers(self, active_server_ids: list) -> None:
+        """
+        Remove tracking for servers that no longer exist.
+
+        Args:
+            active_server_ids: List of currently active server IDs
+        """
+        servers_to_remove = []
+        for server_id in self.server_graphs:
+            if server_id not in active_server_ids:
+                servers_to_remove.append(server_id)
+
+        for server_id in servers_to_remove:
+            self.remove_server(server_id)
+
+        if servers_to_remove:
+            logger.info(f"Cleaned up metrics for {len(servers_to_remove)} inactive servers")
+
+    def get_summary(self) -> Dict[str, Any]:
+        """
+        Get summary of all tracked servers.
+
+        Returns:
+            Dictionary with tracking statistics
+        """
+        return {
+            "total_servers": len(self.server_graphs),
+            "servers_with_data": sum(
+                1 for graphs in self.server_graphs.values() if graphs.has_sufficient_data
+            ),
+            "total_data_points": sum(
+                len(graphs.data_points) for graphs in self.server_graphs.values()
+            ),
+        }

models.py

@@ -0,0 +1,112 @@
+"""
+Data Models Module
+
+Pydantic models for type-safe data handling throughout the application.
+Provides validation and serialization for API responses and internal data.
+"""
+
+from typing import Any, Dict, List, Optional
+from pydantic import BaseModel, Field
+
+
+class ResourceAttributes(BaseModel):
+    """Current resource usage attributes from panel API."""
+
+    current_state: str
+    resources: "ResourceUsage"
+
+
+class ResourceUsage(BaseModel):
+    """Individual resource metrics."""
+
+    cpu_absolute: float = 0.0
+    memory_bytes: int = 0
+    disk_bytes: int = 0
+    network_rx_bytes: int = 0
+    network_tx_bytes: int = 0
+    uptime: int = 0
+
+
+class ServerAttributes(BaseModel):
+    """Server attributes from panel API."""
+
+    identifier: str
+    name: str
+    description: Optional[str] = None
+    suspended: bool = False
+    limits: "ServerLimits" = Field(default_factory=lambda: ServerLimits())
+
+
+class ServerLimits(BaseModel):
+    """Server resource limits."""
+
+    cpu: int = 0
+    memory: int = 0
+    disk: int = 0
+
+
+class ServerRelationships(BaseModel):
+    """Server relationships data."""
+
+    allocations: Optional["AllocationsData"] = None
+
+
+class AllocationsData(BaseModel):
+    """Allocations data container."""
+
+    data: List["Allocation"] = Field(default_factory=list)
+
+
+class Allocation(BaseModel):
+    """Individual allocation (IP/port)."""
+
+    attributes: "AllocationAttributes"
+
+
+class AllocationAttributes(BaseModel):
+    """Allocation attributes."""
+
+    ip_alias: str = "Unknown"
+    port: int = 0
+    is_default: bool = False
+
+
+class ServerData(BaseModel):
+    """Complete server data from panel API."""
+
+    attributes: ServerAttributes
+    relationships: Optional[ServerRelationships] = None
+
+
+class ServerResources(BaseModel):
+    """Server resources response from panel API."""
+
+    attributes: ResourceAttributes
+
+
+class APIResponse(BaseModel):
+    """Generic API response wrapper."""
+
+    status: str = "success"
+    data: Any = None
+    message: Optional[str] = None
+    errors: Optional[List[Dict[str, Any]]] = None
+
+
+class ServerListResponse(BaseModel):
+    """Server list response from panel API."""
+
+    data: List[Dict[str, Any]] = Field(default_factory=list)
+    meta: Optional[Dict[str, Any]] = None
+
+
+class EmbedLocation(BaseModel):
+    """Embed location tracking data."""
+
+    channel_id: str
+    message_id: str
+
+
+# Rebuild models to resolve forward references
+ResourceAttributes.model_rebuild()
+ServerAttributes.model_rebuild()

requirements.txt

@@ -3,3 +3,4 @@ aiohttp>=3.14.5
 configparser>=7.3.0
 python-dotenv>=1.2.4
 matplotlib>=3.11.2
+pydantic>=2.14.1

views.py

@@ -0,0 +1,315 @@
+"""
+Discord Views Module
+
+Contains all Discord UI views and interactive components.
+Handles server control buttons, dropdowns, and permission checks.
+"""
+
+import logging
+from typing import TYPE_CHECKING
+
+import discord
+from discord.ext import commands
+
+from config import Config
+from exceptions import PermissionError
+
+if TYPE_CHECKING:
+    from api import PanelAPI
+
+logger = logging.getLogger("pterodisbot")
+
+
+class ServerStatusView(discord.ui.View):
+    """
+    Interactive view containing server control buttons.
+
+    Provides persistent controls for server management with role-based access.
+    Buttons: Start (green), Stop (red), Restart (blurple), Show Address (grey).
+
+    Features:
+    - Guild restriction (only works in configured server)
+    - Role-based access control (requires "Game Server User" role)
+    - Persistent views (timeout=None for infinite persistence)
+    - Comprehensive error handling with user feedback
+    """
+
+    def __init__(
+        self,
+        server_id: str,
+        server_name: str,
+        api: "PanelAPI",
+        server_data: dict,
+    ) -> None:
+        """
+        Initialize the server status view with control buttons.
+
+        Args:
+            server_id: The server's panel identifier
+            server_name: Human-readable server name
+            api: Panel API client instance
+            server_data: Full server data from panel
+        """
+        super().__init__(timeout=None)  # Persistent view
+        self.server_id = server_id
+        self.server_name = server_name
+        self.api = api
+        self.server_data = server_data
+        logger.debug(f"Created ServerStatusView for {server_name} ({server_id})")
+
+    async def interaction_check(self, interaction: discord.Interaction) -> bool:
+        """
+        Verify the interacting user has the required role and is in the allowed guild.
+
+        Args:
+            interaction: Discord interaction object
+
+        Returns:
+            bool: True if authorized, False otherwise
+        """
+        config = Config.get_instance()
+
+        # First check if interaction is from the allowed guild
+        if interaction.guild_id != config.allowed_guild_id:
+            logger.warning(f"Unauthorized interaction attempt from guild {interaction.guild_id}")
+            await interaction.response.send_message(
+                "This bot is only available in a specific server.", ephemeral=True
+            )
+            return False
+
+        # Then check for required role
+        logger.debug(f"Checking permissions for {interaction.user.name} on server {self.server_name}")
+        has_role = any(role.name == config.required_role for role in interaction.user.roles)
+        if not has_role:
+            logger.warning(
+                f"Permission denied for {interaction.user.name} - missing '{config.required_role}' role"
+            )
+            await interaction.response.send_message(
+                f"You don't have permission to control servers. You need the '{config.required_role}' role.",
+                ephemeral=True,
+            )
+            return False
+
+        logger.debug(f"Permission granted for {interaction.user.name}")
+        return True
+
+    async def on_error(
+        self, interaction: discord.Interaction, error: Exception, item: discord.ui.Item
+    ) -> None:
+        """
+        Handle errors in button interactions.
+
+        Args:
+            interaction: Discord interaction object
+            error: Exception that occurred
+            item: The UI item that triggered the error
+        """
+        logger.error(f"View error in {self.server_name} by {interaction.user.name}: {str(error)}")
+        if not interaction.response.is_done():
+            await interaction.response.send_message(
+                "An error occurred while processing your request.", ephemeral=True
+            )
+        else:
+            await interaction.followup.send(
+                "An error occurred while processing your request.", ephemeral=True
+            )
+
+    @discord.ui.button(
+        label="Start", style=discord.ButtonStyle.green, custom_id="start_button"
+    )
+    async def start_button(
+        self, interaction: discord.Interaction, button: discord.ui.Button
+    ) -> None:
+        """Send a start command to the server."""
+        logger.info(f"Start button pressed for {self.server_name} by {interaction.user.name}")
+        await interaction.response.defer(ephemeral=True)
+        result = await self.api.send_power_action(self.server_id, "start")
+
+        if result.get("status") == "success":
+            message = f"Server '{self.server_name}' is starting..."
+            logger.info(f"Successfully started server {self.server_name}")
+        else:
+            message = f"Failed to start server: {result.get('message', 'Unknown error')}"
+            logger.error(f"Failed to start server {self.server_name}: {message}")
+
+        await interaction.followup.send(message, ephemeral=True)
+
+    @discord.ui.button(
+        label="Stop", style=discord.ButtonStyle.red, custom_id="stop_button"
+    )
+    async def stop_button(
+        self, interaction: discord.Interaction, button: discord.ui.Button
+    ) -> None:
+        """Send a stop command to the server."""
+        logger.info(f"Stop button pressed for {self.server_name} by {interaction.user.name}")
+        await interaction.response.defer(ephemeral=True)
+        result = await self.api.send_power_action(self.server_id, "stop")
+
+        if result.get("status") == "success":
+            message = f"Server '{self.server_name}' is stopping..."
+            logger.info(f"Successfully stopped server {self.server_name}")
+        else:
+            message = f"Failed to stop server: {result.get('message', 'Unknown error')}"
+            logger.error(f"Failed to stop server {self.server_name}: {message}")
+
+        await interaction.followup.send(message, ephemeral=True)
+
+    @discord.ui.button(
+        label="Restart", style=discord.ButtonStyle.blurple, custom_id="restart_button"
+    )
+    async def restart_button(
+        self, interaction: discord.Interaction, button: discord.ui.Button
+    ) -> None:
+        """Send a restart command to the server."""
+        logger.info(f"Restart button pressed for {self.server_name} by {interaction.user.name}")
+        await interaction.response.defer(ephemeral=True)
+        result = await self.api.send_power_action(self.server_id, "restart")
+
+        if result.get("status") == "success":
+            message = f"Server '{self.server_name}' is restarting..."
+            logger.info(f"Successfully restarted server {self.server_name}")
+        else:
+            message = f"Failed to restart server: {result.get('message', 'Unknown error')}"
+            logger.error(f"Failed to restart server {self.server_name}: {message}")
+
+        await interaction.followup.send(message, ephemeral=True)
+
+    @discord.ui.button(
+        label="Show Address",
+        style=discord.ButtonStyle.grey,
+        custom_id="show_address_button",
+    )
+    async def show_address_button(
+        self, interaction: discord.Interaction, button: discord.ui.Button
+    ) -> None:
+        """Show server's default allocation IP and port."""
+        logger.info(f"Show Address button pressed for {self.server_name} by {interaction.user.name}")
+        try:
+            await interaction.response.defer(ephemeral=True)
+            logger.debug(f"Fetching server details for {self.server_id}")
+
+            ip_alias, port = await self.api.get_server_default_allocation(self.server_id)
+
+            # Create and send embed
+            embed = discord.Embed(
+                title=f"{self.server_name} Connection Info",
+                color=discord.Color.blue(),
+                description=f"Server ID: `{self.server_id}`",
+            )
+            embed.add_field(name="Address", value=f"`{ip_alias}`", inline=True)
+            embed.add_field(name="Port", value=f"`{port}`", inline=True)
+
+            await interaction.followup.send(embed=embed, ephemeral=True)
+            logger.info(f"Displayed connection info for {self.server_name}")
+
+        except Exception as e:
+            logger.error(f"Failed to show address for {self.server_name}: {str(e)}")
+            await interaction.followup.send(
+                "⚠️ Failed to get connection info. The server may not have any ports allocated.",
+                ephemeral=True,
+            )
+
+
+class ServerDropdown(discord.ui.Select):
+    """Dropdown menu for server selection in status dashboard."""
+
+    def __init__(self, server_options: list) -> None:
+        """
+        Initialize the server dropdown.
+
+        Args:
+            server_options: List of discord.SelectOption objects
+        """
+        super().__init__(
+            placeholder="Select a server to display...",
+            options=server_options,
+            min_values=1,
+            max_values=1,
+        )
+
+    async def callback(self, interaction: discord.Interaction) -> None:
+        """
+        Handle server selection from dropdown menu.
+
+        Creates permanent status embed in the channel for the selected server.
+        """
+        from bot import PterodactylBot
+
+        await interaction.response.defer(ephemeral=True)
+
+        bot = PterodactylBot.get_instance()
+        selected_server_id = self.values[0]
+        server_data = bot.server_cache.get(selected_server_id)
+
+        if not server_data:
+            await interaction.followup.send(
+                "❌ Selected server no longer available. Please try again.", ephemeral=True
+            )
+            return
+
+        server_name = server_data["attributes"]["name"]
+        logger.info(f"User {interaction.user.name} selected server: {server_name}")
+
+        try:
+            # Get current server status for embed creation
+            resources = await bot.api.get_server_resources(selected_server_id)
+
+            # Delete old embed if it exists to prevent duplication
+            if selected_server_id in bot.embed_manager.embed_locations:
+                logger.debug(f"Found existing embed for {selected_server_id}, attempting to delete")
+                try:
+                    old_location = bot.embed_manager.embed_locations[selected_server_id]
+                    old_channel = bot.get_channel(int(old_location["channel_id"]))
+                    if old_channel:
+                        try:
+                            old_message = await old_channel.fetch_message(
+                                int(old_location["message_id"])
+                            )
+                            await old_message.delete()
+                            logger.debug(f"Deleted old embed for {selected_server_id}")
+                        except discord.NotFound:
+                            logger.debug(f"Old embed for {selected_server_id} already deleted")
+                except Exception as e:
+                    logger.error(f"Failed to delete old embed: {str(e)}")
+
+            # Create and send new permanent status embed in channel
+            from models import ServerData, ServerResources
+
+            server_obj = ServerData.model_validate(server_data)
+            resources_obj = ServerResources.model_validate(resources)
+
+            embed, view = await bot.embed_manager.create_server_embed(server_obj, resources_obj)
+            message = await interaction.channel.send(embed=embed, view=view)
+            await bot.embed_manager.track_embed(selected_server_id, message)
+
+            await interaction.followup.send(
+                f"✅ **{server_name}** status has been posted in {interaction.channel.mention}",
+                ephemeral=True,
+            )
+            logger.info(f"Successfully posted status for {server_name}")
+
+        except Exception as e:
+            logger.error(f"Failed to create status embed for {server_name}: {str(e)}")
+            await interaction.followup.send(
+                f"❌ Failed to create status embed for **{server_name}**: {str(e)}", ephemeral=True
+            )
+
+
+class ServerDropdownView(discord.ui.View):
+    """View containing the server dropdown with timeout for auto-cleanup."""
+
+    def __init__(self, server_options: list, timeout: int = 180) -> None:
+        """
+        Initialize dropdown view with timeout.
+
+        Args:
+            server_options: List of discord.SelectOption objects
+            timeout: Timeout in seconds (default 3 minutes for ephemeral cleanup)
+        """
+        super().__init__(timeout=timeout)
+        self.server_options = server_options
+        self.add_item(ServerDropdown(server_options))
+
+    async def on_timeout(self) -> None:
+        """Clean up when dropdown times out (ephemeral auto-removal)."""
+        logger.debug("Server dropdown timed out and was automatically cleaned up")