servii-backend/generic_executor.py

from http import HTTPStatus

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, str or 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 is None:
                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, 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]
        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 account_create(user: UserRecord) -> tuple[HTTPStatus, str or 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, 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.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)
        file_manager.update_server_property(server_path + "/server.properties", "server-port", port)
        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, 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 as e:
        file_manager.log_error(type(e).__name__, str(e))
        return HTTPStatus.INTERNAL_SERVER_ERROR, None


def account_delete(user: UserRecord) -> tuple[HTTPStatus, str or 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, str or None]:
    user_id = user.uid
    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)}.")
        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, str or None]:
    port = 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."
        exists: bool = mc_manager.stop_server(port)
        if exists:
            firebase_manager.update_server_running_state(user.uid, 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)
            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, str or 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, str or 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, errors
    return HTTPStatus.OK, f"Successfully updated server {name}."


def run_command(user: UserRecord, command: str, name: str) -> tuple[HTTPStatus, str or 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)
        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