blitz_api/app/apps/service.py

import asyncio

from fastapi import HTTPException, status
from loguru import logger

from app.api.config import config
from app.api.error_report.report import Report
from app.api.models import ApiErrors, ErrorMessage
from app.api.task_utils import get_lock_status
from app.apps.cache import cache as app_cache
from app.apps.constants import AppsServiceKeys, InstallMode
from app.apps.models import AppId, AppStatus, AppStatusQueryResult, AppUninstallInput
from app.apps.tasks import manage_app_task, update_app_state_task
from app.external.result_type.src.result import Err, Ok
from app.system.models import APIPlatform

PLATFORM = config("BAPI_PLATFORM", default=APIPlatform.UNKNOWN)
if PLATFORM == APIPlatform.RASPIBLITZ:
    from app.apps.impl.raspiblitz import RaspiBlitzApps as Apps
elif PLATFORM == APIPlatform.NATIVE_PYTHON:
    from app.apps.impl.native_python import NativePythonApps as Apps
else:
    raise RuntimeError(
        f"Unsupported platform '{PLATFORM}'. Options: {APIPlatform.values_as_list()}."
    )


apps = Apps()

if apps is None:
    raise RuntimeError(f"Unknown platform {PLATFORM}")


async def update_app_state_cache():
    """Starts an update task to the app status cache."""
    update_app_state_task.delay()  # type: ignore


async def get_app_status_single(id: str) -> AppStatus:
    match await apps.get_app_status_single(id):
        case Ok(value):
            return value
        case Err(report):
            # will throw for us
            _handle_error(report)

    # In theory we should never arrive to this point...
    raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Unknown error")


async def get_app_status() -> AppStatusQueryResult:
    match await app_cache.get_cached_app_status():
        case Ok(values) if values:
            return values
        case Ok(_):
            # query executed, but no data was returned
            # This means the cache is empty or stale => trigger update
            update_app_state_task.delay()  # type: ignore
            raise HTTPException(
                status.HTTP_500_INTERNAL_SERVER_ERROR,
                detail="Cached app status is empty. App status update task triggered.",
            )
        case Err(report):
            _handle_error(report)

    raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Unknown error")


async def get_app_status_advanced(app_id: str) -> AppStatus:
    match await apps.get_app_status_advanced(app_id):
        case Ok(values):
            return values
        case Err(report):
            _handle_error(report)

    raise HTTPException(status.HTTP_500_INTERNAL_SERVER_ERROR, detail="Unknown error")


async def install_app(app_id: AppId):
    res = await get_lock_status(AppsServiceKeys.APP_MANAGE_LOCK_KEY)
    match res:
        case Ok(value):
            if value:
                raise HTTPException(
                    status.HTTP_423_LOCKED,
                    detail="App install or uninstall task already running",
                )
        case Err(report):
            _handle_error(report)

    manage_app_task.delay(app_id, InstallMode.ON)  # type: ignore

    asyncio.create_task(_watcher(app_id, "install"))


async def uninstall_app(data: AppUninstallInput):
    res = await get_lock_status(AppsServiceKeys.APP_MANAGE_LOCK_KEY)
    match res:
        case Ok(value):
            if value:
                raise HTTPException(
                    status.HTTP_423_LOCKED,
                    detail="App install or uninstall task already running",
                )
        case Err(report):
            _handle_error(report)

    app_id = data.app_id

    manage_app_task.delay(app_id, InstallMode.OFF, data.keep_data)  # type: ignore

    asyncio.create_task(_watcher(app_id, "uninstall"))


async def _watcher(app_id: AppId, action: str):
    from app.apps.tasks_impl.listeners import AppManageListener

    logger.info(f"Starting app {action} listener for app {app_id}")
    listener = AppManageListener()
    await listener.connect()
    await listener.listen()
    logger.info(f"Closing app {action} listener for app {app_id}")
    logger.info("Triggering app state cache update")
    # TODO: Not sure if this is the right place to trigger the cache update
    #       or if it should be done by the API at all
    #       problem is that it's also triggered when no action was taken
    #       (e.g. when the app is already installed)
    await update_app_state_cache()


def _handle_error(report: Report):
    code = 0
    if isinstance(report.last_error, HTTPException):
        code = report.last_error.status_code
        if code >= 500:
            # internal error, log fully
            logger.error(report.format_verbose(include_sensitive=True))
        else:
            # user error, log only as debug
            logger.info(report.format_verbose())

        raise HTTPException(
            status_code=code,
            detail=ErrorMessage(
                detail=report.frames[0].message,
                error_code=ApiErrors.INVALID_REQUEST_INPUT,
                report=f"{report.format_verbose()}",
            ).model_dump(),
        )
    else:
        logger.error(
            f"Error report didn't contain a HTTPException as root_error: "
            f"{type(report.root_error)}\n"
            f"{report.format()}"
        )

        raise HTTPException(
            status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=ErrorMessage(
                detail=f"{report.format()}",
            ).model_dump(),
        )