servii-backend/generic_executor.py

import json
import os
from http import HTTPStatus
from typing import Union, Tuple, Dict

from cloudflare.types.dns import SRVRecord
from firebase_admin.auth import UserRecord

import cloudflare_manager
import file_manager
import firebase_manager
from server_mc_manager import MinecraftServerManager

mc_manager: MinecraftServerManager = MinecraftServerManager()


def set_subdomain(user: UserRecord, subdomain: str) -> tuple[HTTPStatus, Union[str, None]]:
    user_id: str = user.uid
    store = firebase_manager.firestore_database
    _users = store.collection("users")
    query = _users.where(field_path="subdomain", op_string="==", value=subdomain)
    for _ in query.stream():
        return HTTPStatus.ALREADY_REPORTED, "Subdomain already associated."
    try:
        if firebase_manager.get_user_field(user_id, "subdomain"):
            return HTTPStatus.OK, "You cannot change your subdomain."
        exists: bool = firebase_manager.update_firestore(user_id, {'subdomain': subdomain})
        if not exists:
            account_create(user)
            return set_subdomain(user, subdomain)
        port = firebase_manager.get_server_port(user_id)
        if port is None:
            return HTTPStatus.NOT_FOUND, "Port not found on database"
        try:
            record: SRVRecord = cloudflare_manager.add_cloudflare_dns(subdomain, port)
            record_id: str = record.id
            if record_id == "":
                return HTTPStatus.BAD_REQUEST, "Failed to add record to cloudflare"
            firebase_manager.update_firestore(user_id, {'cloudflare_entry': str(record_id)})
        except Exception as e:
            firebase_manager.update_firestore(user_id, {'subdomain': None})
            return HTTPStatus.INTERNAL_SERVER_ERROR, f"Cloudflare configuration failed. {str(e)}"
        return HTTPStatus.OK, "Successfully associated subdomain."
    except Exception as e:
        return HTTPStatus.FORBIDDEN, str(e)


def fetch_servers(user: UserRecord) -> tuple[HTTPStatus, Union[str, list]]:
    user_id: str = user.uid
    server_ref = firebase_manager.firestore_database.collection('users').document(user_id).collection('servers')
    try:
        servers = list(server_ref.stream())
        servers_data: list[dict[str, any]] = []
        for doc in servers:
            server_info = doc.to_dict()
            if server_info.get('running', False):
                try:
                    connected_players = mc_manager.get_online_players(server_info.get('port', 25565))
                except Exception as e:
                    file_manager.log_error(type(e).__name__, str(e))
                    connected_players = 0
                server_info['onlinePlayers'] = connected_players
            servers_data.append(server_info)
        if not servers_data:
            account_create(user)
        return HTTPStatus.OK, servers_data
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return account_create(user)


def fetch_logs(user: UserRecord, name: str) -> tuple[HTTPStatus, Union[str, None]]:
    user_id: str = user.uid
    lines_to_read: int = 200
    log_file = f"users/{user_id}/{name}/logs/latest.log"
    try:
        with open(log_file, "r") as f:
            logs = f.readlines()[-lines_to_read:]
            return HTTPStatus.OK, str(logs)
    except FileNotFoundError:
        return HTTPStatus.NOT_FOUND, "Log file not found."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))


def fetch_history(user: UserRecord, name: str) -> tuple[HTTPStatus, Union[str, None]]:
    user_id: str = user.uid
    history = f"users/{user_id}/{name}/history.log"
    try:
        with open(history, "r") as f:
            logs = f.readlines()
            return HTTPStatus.OK, str(logs)
    except FileNotFoundError:
        return HTTPStatus.NOT_FOUND, "History file not found."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, "Unknown error."


def fetch_players_status(user: UserRecord, name: str) -> tuple[HTTPStatus, Union[dict[str, str], str]]:
    user_id: str = user.uid
    server_path = f"users/{user_id}/{name}"
    players_status_files = {
        "Whitelist": "whitelist.json",
        "BannedPlayers": "banned-players.json",
        "BannedIps": "banned-ips.json",
        "Operators": "ops.json"
    }
    try:
        file_contents = {}

        for file_name, file_path in players_status_files.items():
            full_file_path = f"{server_path}/{file_path}"
            if os.path.exists(full_file_path):
                with open(full_file_path, 'r') as file:
                    file_contents[file_name] = file.read()
            else:
                return HTTPStatus.NOT_FOUND, "Launch the server at least once."

        file_contents_json = json.dumps(file_contents, indent=4)

        return HTTPStatus.OK, file_contents
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, "Unknown error."


def account_create(user: UserRecord) -> tuple[HTTPStatus, Union[str, None]]:
    if firebase_manager.user_field_exists(user.uid):
        return HTTPStatus.FORBIDDEN, "User already exists."
    try:
        file_manager.create_folder("users/" + user.uid)
        firebase_manager.set_firestore(user.uid, {'mail': user.email,
                                                  'name': user.display_name,
                                                  'photo_url': user.photo_url,
                                                  'subdomain': None,
                                                  'port': firebase_manager.fetch_port() + 1,
                                                  'cloudflare_entry': None,
                                                  })
        return HTTPStatus.CREATED, "Successfully created account."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, "Error creating account."


allowed_frameworks: list[str] = ["paper", "spigot", "bukkit"]


def server_create(user: UserRecord, name: str, version: str, framework: str = "paper") -> (
        tuple)[HTTPStatus, Union[str, None]]:
    if framework not in allowed_frameworks:
        return HTTPStatus.METHOD_NOT_ALLOWED, f"Framework {framework} not recognized."
    user_id = user.uid
    server_path: str = f"users/{user_id}/{name}"
    server_template_path: str = f"servers/{framework}/{version}"
    try:
        port = firebase_manager.get_server_port(user_id)
        if port is None:
            return HTTPStatus.NOT_FOUND, "Port not found on database"
        subdomain = firebase_manager.firestore_database.collection("users").document(user_id).get().get("subdomain")
        if subdomain is None:
            firebase_manager.delete_server(user_id, name)
            return HTTPStatus.NOT_FOUND, f"You haven't associated a subdomain yet."
        if firebase_manager.server_name_taken(user_id, name):
            return HTTPStatus.CONFLICT, f"Server name '{name}' already in use."
        firebase_manager.create_server(user_id, name, version, port, framework)
        file_manager.create_folder(server_path)
        file_manager.copy_folder_contents(server_template_path, server_path)
        file_manager.copy_folder_contents("servers/shared", server_path)
        prop_path: str = server_path + "/server.properties"
        file_manager.update_server_property(prop_path, "server-port", port)
        file_manager.update_server_property(prop_path, "query.port", port)
        file_manager.update_server_property(prop_path, "enable-query", "true")
        file_manager.log_action(user_id, name, "ServerCreate")
        return HTTPStatus.CREATED, f"Successfully created server '{name}'."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Server creation failed | {e}"


def server_delete(name: str, user: UserRecord) -> tuple[HTTPStatus, Union[str, None]]:
    user_id = user.uid
    try:
        port: int = firebase_manager.get_server_port(user_id)
        mc_manager.stop_server_forcefully(port)
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e)+f" error when stopping server {name}")
    server_path: str = f"users/{user_id}/{name}"
    try:
        firebase_manager.delete_server(user_id, name)
        file_manager.delete_non_empty_folder(server_path)
        return HTTPStatus.OK, f"Successfully deleted server '{name}'."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, None


def account_delete(user: UserRecord) -> tuple[HTTPStatus, Union[str, None]]:
    user_id = user.uid
    try:
        dns_record_id = firebase_manager.get_user_field(user_id, "cloudflare_entry")
        if dns_record_id is None:
            return HTTPStatus.IM_A_TEAPOT, "Cloudflare 'dns_record_id' not found in Firestore."
        cloudflare_manager.delete_cloudflare_dns(dns_record_id)
    except Exception as e:
        return HTTPStatus.NOT_MODIFIED, f"Account deletion failed. {str(e)}"
    try:
        firebase_manager.delete_user(user_id)
    except Exception as e:
        return HTTPStatus.EXPECTATION_FAILED, f"Database deletion failed | {e}"
    try:
        file_manager.delete_non_empty_folder("users/" + user_id)
        return HTTPStatus.OK, f"Successfully deleted user '{user_id}'."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error deleting user '{user_id}' on the server."


def server_run(user: UserRecord, name: str) -> tuple[HTTPStatus, Union[str, None]]:
    user_id = user.uid
    mc_manager.set_cooldown(user_id=user_id)
    try:
        port: int = firebase_manager.get_server_port(user_id)
        server_id = mc_manager.start_server(f"users/{user_id}/{name}", port)
        if server_id is None:
            return HTTPStatus.OK, f"You cannot run multiples instances at this time."
        mc_manager.servers[server_id]['port'] = int(port)
        try:
            firebase_manager.update_server_running_state(user_id, name, True)
        except Exception as e:
            return HTTPStatus.INTERNAL_SERVER_ERROR, (f"Error updating server {name}'s running state in database. "
                                                      f"{type(e).__name__}{str(e)}.")
        file_manager.log_action(user_id, name, "ServerRun")
        return HTTPStatus.ACCEPTED, f"Successfully started server '{name}'."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error when running server: {e}"


def server_stop(user: UserRecord, name: str) -> tuple[HTTPStatus, Union[str, None]]:
    port = None
    user_id: str = user.uid
    if mc_manager.has_cooldown(user_id=user_id):
        return HTTPStatus.FORBIDDEN, f"Wait 30 seconds before stopping '{name}'."
    mc_manager.set_cooldown(user_id=user_id)
    try:
        port = firebase_manager.get_server_port(user_id)
        if port is None:
            return HTTPStatus.NOT_FOUND, f"Server '{name}' not found in firestore."
        exists: bool = mc_manager.stop_server(port)
        if exists:
            firebase_manager.update_server_running_state(user_id, name, False)
            return HTTPStatus.OK, f"Successfully stopped server '{name}'."
        return HTTPStatus.OK, f"Server '{name}' already stopped."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        if port:
            mc_manager.stop_server_forcefully(port)
            firebase_manager.update_server_running_state(user_id, name, False)
            file_manager.log_action(user_id, name, "ServerStop")
            return HTTPStatus.OK, f"Successfully stopped server '{name}'."
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error occurred when stopping server '{name}'."


def update_property(uid: str, name: str, prop: str, value: str) -> tuple[HTTPStatus, Union[str, None]]:
    property_file_path: str = f"users/{uid}/{name}/server.properties"
    try:
        file_manager.update_server_property(property_file_path, prop, value)
        firebase_manager.update_server_property(uid, name, prop, value)
        return HTTPStatus.OK, f"Successfully set '{prop}' to '{value}'."
    except ValueError as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.BAD_REQUEST, f"Property '{prop}' not found."
    except FileNotFoundError:
        return HTTPStatus.NOT_FOUND, f"File server.properties for server '{name}' not found."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Unhandled error: {type(e).__name__}, {str(e)}"


def update_properties(user: UserRecord, name: str, props: list[tuple[str, str]]) -> tuple[HTTPStatus, Union[str, None]]:
    errors: list[str] = []
    for prop, value in props:
        if prop not in MinecraftServerManager.allowed_properties:
            return HTTPStatus.FORBIDDEN, f"Property '{prop}' not allowed."
        status, message = update_property(uid=user.uid, name=name, prop=prop, value=value)
        if status != HTTPStatus.OK:
            errors.append(message)
    if len(errors) > 0:
        return HTTPStatus.IM_A_TEAPOT, str(errors)
    file_manager.log_action(user.uid, name, "UpdateProperties", str(props))
    return HTTPStatus.OK, f"Successfully updated server '{name}'."


def run_command(user: UserRecord, command: str, name: str) -> tuple[HTTPStatus, Union[str, None]]:
    try:
        port = firebase_manager.get_server_port(user.uid)
        if port is None:
            return HTTPStatus.NOT_FOUND, f"Server '{name}' not found in firestore."
        mc_manager.execute_server_command(port, command)
        file_manager.log_action(user.uid, name, "Command", command)
        return HTTPStatus.OK, f"Command '{command}' executed successfully."
    except Exception as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error executing command: {command} || {str(e)}"


if __name__ == "__main__":
    pass