servii-backend/generic_executor.py

from http import HTTPStatus

from firebase_admin.auth import UserRecord

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, str or None]:
    store = firebase_manager.firestore_database
    _users = store.collection("users")
    query = _users.where("subdomain", "==", subdomain)
    for _ in query.stream():
        return HTTPStatus.ALREADY_REPORTED, "Subdomain already associated."
    try:
        exists: bool = firebase_manager.update_firestore(user.uid, {'subdomain': subdomain})
        if not exists:
            account_create(user)
            return set_subdomain(user, subdomain)
        return HTTPStatus.OK, "Successfully associated subdomain."
    except Exception as e:
        return HTTPStatus.FORBIDDEN, str(e)


def fetch_servers(user: UserRecord) -> tuple[HTTPStatus, str or 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 = [doc.to_dict() for doc in servers]
        return HTTPStatus.OK, servers_data
    except Exception as e:
        return account_create(user)


def account_create(user: UserRecord) -> tuple[HTTPStatus, str or None]:
    try:
        file_manager.create_folder("users/" + str(25444))
        firebase_manager.set_firestore(user.uid, {'mail': user.email,
                                                  'name': user.display_name,
                                                  'photo_url': user.photo_url,
                                                  'subdomain': None
                                                  })
        return HTTPStatus.CREATED, "Successfully created account."
    except Exception as 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, str or 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.fetch_port() + 1
        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)
        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)
        file_manager.update_server_property(server_path + "/server.properties", "server-port", port)
        return HTTPStatus.CREATED, f"Successfully created server '{name}'."
    except Exception as e:
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Server creation failed | {e}"


def server_delete(name: str, user: UserRecord) -> tuple[HTTPStatus, str or None]:
    user_id = user.uid
    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:
        return HTTPStatus.INTERNAL_SERVER_ERROR, None


def account_delete(user: UserRecord) -> tuple[HTTPStatus, str or None]:
    user_id = user.uid
    try:
        file_manager.delete_non_empty_folder("users/" + user_id)
        firebase_manager.delete_user(user_id)
        return HTTPStatus.OK, f"Successfully deleted user {user_id}."
    except Exception as e:
        print(f"Error deleting account: {e}")
        return HTTPStatus.INTERNAL_SERVER_ERROR, "Error deleting account."


def server_run(user: UserRecord, name: str) -> tuple[HTTPStatus, str or None]:
    user_id = user.uid
    try:
        port: int = firebase_manager.get_server_port(user_id, name)
        server_id = mc_manager.start_server(f"users/{user_id}/{name}", port)
        mc_manager.servers[server_id]['port'] = int(port)
        return HTTPStatus.ACCEPTED, f"Successfully started server {name}."
    except Exception as e:
        print(f"Error starting server: {e}")
        return HTTPStatus.INTERNAL_SERVER_ERROR, None


def server_stop(user: UserRecord, name: str) -> tuple[HTTPStatus, str or None]:
    port = None
    try:
        port = firebase_manager.get_server_port(user.uid, name)
        if port is None:
            return HTTPStatus.NOT_FOUND, f"Server {name} not found in firestore."
        mc_manager.stop_server(port)
        return HTTPStatus.OK, f"Successfully stopped server {name}."
    except Exception as e:
        if port:
            mc_manager.stop_server_forcefully(port)
        return HTTPStatus.INTERNAL_SERVER_ERROR, f"Error occurred when stopping server {name}."


def update_property(user: UserRecord, name: str, prop: str, value: str) -> tuple[HTTPStatus, str or None]:
    property_file_path: str = f"users/{user.uid}/{name}/server.properties"
    try:
        file_manager.update_server_property(property_file_path, 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 Exception as e:
        print(f"Unhandled error: {type(e).__name__}, {str(e)}")
        return HTTPStatus.INTERNAL_SERVER_ERROR, None


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


if __name__ == "__main__":
    pass