app/backend/fluksio/api/routes/search.py

"""One index of everything in this installation worth jumping to by name."""

from typing import Any, Literal

from fastapi import APIRouter, Depends, Request
from fastapi.concurrency import run_in_threadpool
from pydantic import BaseModel

from fluksio.api.deps import (
    CurrentUser,
    DashboardStoreDep,
    FlowControllerDep,
    get_current_user,
)
from fluksio.api.routes.alerts import read_config as read_alerts_config
from fluksio.flow import modules, panels
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import DashboardNotFound, DashboardStore
from fluksio.flow.secrets import get_secrets
from fluksio.flow.store import FlowNotFound

# A wall panel never reaches this route: ``deps._panel_may`` is a whitelist that
# ends in a 403, and a whole-installation index is the opposite of what a screen
# on a wall is allowed to read.
router = APIRouter(
    prefix="/search", tags=["search"], dependencies=[Depends(get_current_user)]
)

Category = Literal[
    "flow",
    "node",
    "dashboard",
    "widget",
    "panel",
    "secret",
    "module",
    "worker",
    "alert",
]


class SearchEntry(BaseModel):
    """One thing somebody might be looking for.

    Deliberately not a route: where a category lands is the frontend's business,
    and it already owns the router. This says what the thing is and what it is
    called, which is all the matching needs.
    """

    category: Category
    #: The id the frontend routes on.
    name: str
    #: Human title, often empty — a flow is usually only its name.
    title: str = ""
    #: The flow a node sits in, or the dashboard a widget sits on.
    parent: str = ""
    #: Node type, widget type, channel kind.
    kind: str = ""


def _build(
    controller: FlowController, dashboards: DashboardStore, hub: Any, secrets: bool
) -> list[SearchEntry]:
    """Read every store once. Blocking: disk and git throughout.

    # ponytail: rebuilt per call. Key it on ``controller.store.revision`` if a
    # store large enough to feel it ever shows up in a profile.
    """
    entries: list[SearchEntry] = []

    for name in controller.store.list_flows():
        try:
            flow = controller.store.read_flow(name, draft=True)
        except FlowNotFound:
            continue
        entries.append(SearchEntry(category="flow", name=flow.name, title=flow.title))
        entries.extend(
            SearchEntry(
                category="node",
                name=node.id,
                title=node.title,
                parent=flow.name,
                kind=node.type,
            )
            for node in flow.nodes
        )

    for summary in dashboards.list():
        try:
            dashboard = dashboards.read(summary.name, draft=True)
        except DashboardNotFound:
            continue
        entries.append(
            SearchEntry(
                category="dashboard", name=dashboard.name, title=dashboard.title
            )
        )
        entries.extend(
            SearchEntry(
                category="widget",
                name=widget.id,
                title=widget.title,
                parent=dashboard.name,
                kind=widget.type,
            )
            for widget in dashboard.widgets
        )

    entries.extend(
        SearchEntry(category="panel", name=panel.id, title=panel.title)
        for panel in panels.read_config().panels
    )

    if secrets:
        entries.extend(
            SearchEntry(category="secret", name=name) for name in get_secrets().list()
        )

    entries.extend(
        SearchEntry(category="module", name=package.name, kind=package.version)
        for package in modules.info(controller.store).packages
    )

    entries.extend(
        SearchEntry(category="worker", name=worker.name)
        for worker in (hub.workers() if hub is not None else [])
    )

    entries.extend(
        SearchEntry(category="alert", name=channel.name, kind=channel.kind)
        for channel in read_alerts_config().channels
    )

    return entries


@router.get("/", response_model=list[SearchEntry])
async def read_search_index(
    current_user: CurrentUser,
    request: Request,
    controller: FlowControllerDep,
    dashboards: DashboardStoreDep,
) -> Any:
    """Everything searchable, for the client to match against as it is typed.

    The whole index rather than a query: it is a few hundred short rows for an
    installation of any ordinary size, so one fetch when the panel opens beats a
    round trip per keystroke — and the client already has a matcher.

    Secrets are named only to a superuser, which is who ``/secrets`` answers to.
    """
    # Absent when remote workers are switched off — not a reason to fail a search.
    hub = getattr(request.app.state, "worker_hub", None)
    return await run_in_threadpool(
        _build, controller, dashboards, hub, current_user.is_superuser
    )