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

"""The flow API. Everything the editor can do is available here first."""

import asyncio
import time
from typing import Any

import orjson
from fastapi import (
    APIRouter,
    Depends,
    HTTPException,
    Request,
    WebSocket,
    WebSocketDisconnect,
)
from fastapi.concurrency import run_in_threadpool
from jwt.exceptions import InvalidTokenError
from pydantic import BaseModel
from sqlmodel import Session, col, select

from fluksio.api.deps import (
    CurrentUser,
    FlowControllerDep,
    SessionDep,
    decode_token,
    get_current_user,
    user_from_token,
)
from fluksio.core.db import engine
from fluksio.flow.artifacts import is_reference
from fluksio.flow.controller import FlowController
from fluksio.flow.dashboards import DashboardStore
from fluksio.flow.events import event_bus
from fluksio.flow.messages import flow_of, qualify
from fluksio.flow.panels import messages_for
from fluksio.flow.pipeline import ValidationIssue
from fluksio.flow.runs import RunRejected
from fluksio.flow.schemas import (
    NAME_PATTERN,
    BrainGraph,
    FlowDef,
    FlowsPublic,
    FlowStatePublic,
    FlowSummary,
    HistoryPoint,
    LibraryNode,
    MessageHistory,
    MessageValue,
    NodeSource,
    NodeStatusPublic,
    NodeTypeInfo,
)
from fluksio.flow.state import as_number
from fluksio.flow.store import (
    FlowExists,
    FlowNotFound,
    LibExists,
    LibNotFound,
    StaleVersion,
)
from fluksio.models import Flavor, Message, Run

router = APIRouter(
    prefix="/flows", tags=["flows"], dependencies=[Depends(get_current_user)]
)
# The websocket authenticates from its query string, so it stays off the
# router that enforces the HTTP bearer scheme.
ws_router = APIRouter(prefix="/flows", tags=["flows"])


class Endpoint(BaseModel):
    """Something wired into this flow that is not a node in it.

    A dashboard control setting one of its messages, a tile showing one, or a
    node in another flow on the far side of a dotted name. The canvas draws
    these so a value never appears to come from nowhere — or worse, appears to
    come from whichever node happens to be drawn as a producer.
    """

    #: dashboard or flow.
    kind: str
    #: Stable within its kind, and used as the canvas node id.
    id: str
    label: str
    #: What sort of widget, or the node type in the other flow.
    detail: str = ""
    #: Messages of this flow it publishes, and ones it reads.
    provides: list[str] = []
    requires: list[str] = []


class FlowDetail(BaseModel):
    """A flow plus how it is currently doing.

    ``definition`` is the working copy — the unpublished draft when there is
    one — because that is what the editor shows. ``nodes`` reports the
    published flow, which is what is actually running.
    """

    definition: FlowDef
    nodes: list[NodeStatusPublic] = []
    issues: list[ValidationIssue] = []
    has_draft: bool = False
    enabled: bool = True
    paused: bool = False
    #: Dashboards and other flows wired into this one.
    endpoints: list[Endpoint] = []


class ValidationResult(BaseModel):
    issues: list[ValidationIssue] = []


class RenameRequest(BaseModel):
    new_name: str


class PublishRequest(BaseModel):
    version: int


class ShareRequest(BaseModel):
    lib_name: str


class RunRequest(BaseModel):
    inputs: dict[str, Any] = {}


class TriggerRequest(BaseModel):
    values: dict[str, Any] = {}


def _endpoints(controller: FlowController, definition: FlowDef) -> list[Endpoint]:
    """Everything wired into this flow from outside it."""
    flow = definition.name
    found: dict[str, Endpoint] = {}

    dashboards: DashboardStore | None = getattr(controller, "dashboards", None)
    if dashboards is not None:
        for binding in dashboards.bindings_for(flow):
            found[f"dashboard:{binding['dashboard']}:{binding['widget']}"] = Endpoint(
                kind="dashboard",
                id=f"dashboard:{binding['dashboard']}:{binding['widget']}",
                label=binding["title"],
                detail=binding["type"],
                provides=[binding["provides"]] if binding["provides"] else [],
                requires=binding["requires"],
            )

    def _flow_endpoint(key: str, detail: str) -> Endpoint:
        """One label per node on the far side, however many names reach it."""
        return found.setdefault(
            f"flow:{key}",
            Endpoint(kind="flow", id=f"flow:{key}", label=key, detail=detail),
        )

    # An outsider reaching into this flow.
    for other, node_id, provides, requires in controller.cross_flow_nodes(flow):
        endpoint = _flow_endpoint(f"{other}.{node_id}", "flow")
        endpoint.provides += provides
        endpoint.requires += requires

    # And this flow reaching out: its own ports bound to a message of another
    # flow. Read from the working document rather than from the store, so a
    # name just typed is drawn before it has been published — the same reason
    # the canvas draws its own boundary from the document.
    for node in definition.nodes:
        for spec, ours_publishes in [
            *((spec, False) for spec in node.requires),
            *((spec, True) for spec in node.provides),
        ]:
            message = qualify(flow, spec.name or "")
            if not message or flow_of(message) == flow:
                continue
            # The node at the other end, so the label reads like a dashboard's:
            # what it is on the first line, what sort of thing it is on the
            # second. A message no published flow declares yet has no other end
            # to name, so it is drawn as the message it is.
            far = controller.message_node(
                message, published=not ours_publishes, exclude=flow
            )
            endpoint = _flow_endpoint(
                f"{far.flow}.{far.node}" if far else message,
                far.type if far else "flow",
            )
            # An endpoint publishing into this flow is what this flow reads.
            side = endpoint.requires if ours_publishes else endpoint.provides
            if message not in side:
                side.append(message)

    return list(found.values())


def _detail(controller: FlowController, definition: FlowDef) -> FlowDetail:
    name = definition.name
    running = {
        "enabled": controller.is_enabled(name),
        "paused": controller.is_paused(name),
    }
    endpoints = _endpoints(controller, definition)
    if controller.store.has_draft(name):
        # Report the draft the editor is showing, not the version running
        # underneath it — otherwise a node the author just broke looks fine.
        preview = controller.preview(name)
        return FlowDetail(
            definition=definition,
            nodes=preview.nodes,
            issues=preview.issues,
            has_draft=True,
            endpoints=endpoints,
            **running,
        )
    return FlowDetail(
        definition=definition,
        nodes=controller.node_statuses(name),
        issues=controller.flow_issues(name),
        endpoints=endpoints,
        **running,
    )


def _read_flow(controller: FlowController, name: str) -> FlowDef:
    """The working copy: the draft when there is one, else what is published."""
    try:
        return controller.store.read_flow(name, draft=True)
    except FlowNotFound:
        raise HTTPException(status_code=404, detail=f"No flow named '{name}'")


def _audit(action: str, flow: str, user: CurrentUser) -> None:
    """Record who changed what. The collector writes it down; the bus carries it."""
    event_bus.publish(
        {
            "type": "audit",
            "action": action,
            "flow": flow,
            "user": user.email,
            "ts": time.time(),
        }
    )


def _source_ref(definition: FlowDef, node_id: str) -> str | None:
    """The library source this node runs, if it is a shared one."""
    node = next((n for n in definition.nodes if n.id == node_id), None)
    return node.source_ref if node else None


def _require_enabled(controller: FlowController, name: str) -> None:
    if not controller.is_enabled(name):
        raise HTTPException(
            status_code=409,
            detail=f"Flow '{name}' is stopped — start it before running it",
        )


def _flow_state(controller: FlowController, name: str) -> FlowStatePublic:
    return FlowStatePublic(
        values={
            key: MessageValue(**value) for key, value in controller.values(name).items()
        },
        nodes=controller.node_statuses(name),
    )


# -----------------------------------------------------------------------------
# Flows
# -----------------------------------------------------------------------------


@router.get("/", response_model=FlowsPublic)
def read_flows(controller: FlowControllerDep) -> Any:
    """List every flow."""
    summaries = []
    for name in controller.store.list_flows():
        try:
            definition = controller.store.read_flow(name, draft=True)
        except FlowNotFound:
            continue
        statuses = controller.node_statuses(name)
        summaries.append(
            FlowSummary(
                name=definition.name,
                title=definition.title,
                mode=definition.mode,
                node_count=len(definition.nodes),
                error_count=sum(1 for s in statuses if s.status == "error"),
                has_draft=controller.store.has_draft(name),
                enabled=controller.is_enabled(name),
                paused=controller.is_paused(name),
                quarantined=controller.is_quarantined(name),
                version=definition.version,
                updated_at=controller.store.updated_at(name),
            )
        )
    return FlowsPublic(data=summaries, count=len(summaries))


@router.get("/node-types", response_model=list[NodeTypeInfo])
def read_node_types() -> Any:
    """The node types that can be placed on a canvas."""
    from fluksio.flow.controller import node_type_info

    return node_type_info()


# Above "/{name}" for the same reason "library" is: a flow called "graph" would
# otherwise be unreachable.
@router.get("/graph", response_model=BrainGraph)
def read_graph(controller: FlowControllerDep) -> Any:
    """Every flow as one graph, with nodes talking to the same thing merged."""
    return controller.brain_graph()


# -----------------------------------------------------------------------------
# Shared nodes
#
# Declared above the "/{name}" routes: "library" would otherwise be read as a
# flow name.
# -----------------------------------------------------------------------------


@router.get("/library", response_model=list[LibraryNode])
def read_library(controller: FlowControllerDep) -> Any:
    """The node sources shared across flows, and which nodes use each."""
    return [
        LibraryNode(name=name, used_by=controller.store.usages(name))
        for name in controller.store.list_lib()
    ]


@router.delete("/library/{lib_name}", response_model=Message)
async def delete_shared_node(lib_name: str, controller: FlowControllerDep) -> Any:
    """Remove a shared source, as long as no flow still runs it."""
    used_by = controller.store.usages(lib_name)
    if used_by:
        raise HTTPException(
            status_code=409,
            detail=f"'{lib_name}' is still used by {', '.join(used_by)}",
        )
    try:
        await run_in_threadpool(controller.store.delete_lib_source, lib_name)
    except LibNotFound:
        raise HTTPException(status_code=404, detail=f"No shared node '{lib_name}'")
    return Message(message=f"Deleted shared node '{lib_name}'")


@router.get("/{name}", response_model=FlowDetail)
def read_flow(name: str, controller: FlowControllerDep) -> Any:
    """Read one flow, with the state of its nodes."""
    return _detail(controller, _read_flow(controller, name))


def _check_flavors(definition: FlowDef) -> None:
    """Refuse a size nobody stored, here rather than when the node is built.

    Catching it at the save covers the canvas and ``fluksio sync`` in one
    place; a node that only finds out at build time is a red node somebody has
    to go and look at.
    """
    named = {
        node.resources.flavor
        for node in definition.nodes
        if node.resources is not None and node.resources.flavor
    }
    if not named:
        return
    with Session(engine) as session:
        known = set(session.exec(select(Flavor.name)).all())
    unknown = sorted(named - known)
    if unknown:
        raise HTTPException(
            status_code=422,
            detail=f"No flavor named '{unknown[0]}' "
            f"(known: {', '.join(sorted(known)) or 'none'})",
        )


@router.put("/{name}", response_model=FlowDetail)
async def save_flow(
    name: str,
    definition: FlowDef,
    controller: FlowControllerDep,
) -> Any:
    """Save unpublished changes to a flow.

    This writes a draft: the running pipeline keeps the published version until
    someone publishes. ``version`` is the one the editor last saw — a mismatch
    means another client saved in between and answers 409 rather than throwing
    their work away.
    """
    if definition.name != name:
        raise HTTPException(
            status_code=400, detail="The flow name in the body must match the URL"
        )

    duplicates = {n.id for n in definition.nodes}
    if len(duplicates) != len(definition.nodes):
        raise HTTPException(status_code=400, detail="Node names must be unique")

    _check_flavors(definition)

    try:
        stored = await run_in_threadpool(
            controller.store.write_draft, definition, definition.version
        )
    except StaleVersion as exc:
        raise HTTPException(
            status_code=409,
            detail={"message": str(exc), "current_version": exc.current},
        )
    return _detail(controller, stored)


@router.post("/{name}/publish", response_model=FlowDetail)
async def publish_flow(
    name: str,
    body: PublishRequest,
    controller: FlowControllerDep,
    user: CurrentUser,
) -> Any:
    """Deploy the unpublished changes: the engine picks them up from here."""
    _read_flow(controller, name)
    if not controller.store.has_draft(name):
        raise HTTPException(
            status_code=400, detail=f"Flow '{name}' has no unpublished changes"
        )
    try:
        published = await run_in_threadpool(
            controller.store.publish_flow, name, body.version
        )
    except StaleVersion as exc:
        raise HTTPException(
            status_code=409,
            detail={"message": str(exc), "current_version": exc.current},
        )
    _audit("published", name, user)
    await controller.reload_flow(name)
    return _detail(controller, published)


@router.post("/{name}/discard-draft", response_model=FlowDetail)
async def discard_draft(name: str, controller: FlowControllerDep) -> Any:
    """Throw the unpublished changes away and go back to what is running."""
    _read_flow(controller, name)
    if not controller.store.has_draft(name):
        raise HTTPException(
            status_code=400, detail=f"Flow '{name}' has no unpublished changes"
        )
    if not controller.store.is_published(name):
        raise HTTPException(
            status_code=400,
            detail=(
                f"Flow '{name}' has never been published — delete it instead of "
                "discarding it"
            ),
        )
    published = await run_in_threadpool(controller.store.discard_draft, name)
    return _detail(controller, published)


@router.delete("/{name}", response_model=Message)
async def delete_flow(
    name: str, controller: FlowControllerDep, user: CurrentUser, session: SessionDep
) -> Any:
    """Delete a flow and everything in it, except the record of what it ran."""
    live = session.exec(
        select(Run.id).where(
            col(Run.flow) == name, col(Run.status).in_(("running", "queued"))
        )
    ).first()
    if live is not None:
        raise HTTPException(
            status_code=409,
            detail=(
                f"Flow '{name}' has a run in progress ({live}). Cancel it, or "
                "wait for it to finish, before deleting the flow."
            ),
        )
    try:
        await run_in_threadpool(controller.store.delete_flow, name)
    except FlowNotFound:
        raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
    _audit("deleted", name, user)
    # Its files are gone; its values and queued work would otherwise linger.
    # Its runs stay: a run record is a record of what ran, and surviving the
    # flow it belonged to is the point of keeping one. `DELETE /runs?flow=`
    # is what clears them.
    await run_in_threadpool(controller.forget_flow, name)
    await controller.reload_flow(name)
    return Message(message=f"Deleted flow '{name}'")


@router.post("/{name}/rename", response_model=FlowDetail)
async def rename_flow(
    name: str,
    body: RenameRequest,
    controller: FlowControllerDep,
) -> Any:
    """Rename a flow, along with every reference to its messages."""
    if not NAME_PATTERN.match(body.new_name):
        raise HTTPException(
            status_code=400,
            detail=(
                "Use lowercase letters, digits and underscores, starting with a letter"
            ),
        )
    try:
        renamed = await run_in_threadpool(
            controller.store.rename_flow, name, body.new_name
        )
    except FlowNotFound:
        raise HTTPException(status_code=404, detail=f"No flow named '{name}'")
    except FlowExists as exc:
        raise HTTPException(status_code=409, detail=str(exc))

    # The old name is nobody's namespace now; its values would sit there under a
    # flow that no longer exists. They repopulate under the new name on the next
    # run, so this is cleanup rather than a migration.
    await run_in_threadpool(controller.forget_flow, name)
    await controller.reload()
    return _detail(controller, renamed)


# -----------------------------------------------------------------------------
# Node source
# -----------------------------------------------------------------------------


@router.get("/{name}/nodes/{node_id}/source", response_model=NodeSource)
def read_node_source(
    name: str,
    node_id: str,
    controller: FlowControllerDep,
) -> Any:
    """Read a node's Python source, including unpublished edits."""
    definition = _read_flow(controller, name)
    ref = _source_ref(definition, node_id)
    if ref:
        try:
            return NodeSource(code=controller.store.read_lib_source(ref))
        except LibNotFound:
            raise HTTPException(status_code=404, detail=f"No shared node '{ref}'")
    return NodeSource(
        code=controller.store.read_node_source(name, node_id, draft=True),
        missing=not controller.store.has_node_source(name, node_id, draft=True),
    )


@router.put("/{name}/nodes/{node_id}/source", response_model=NodeStatusPublic)
async def save_node_source(
    name: str,
    node_id: str,
    source: NodeSource,
    controller: FlowControllerDep,
) -> Any:
    """Save a node's source as an unpublished edit and report whether it loads.

    The answer comes from compiling the code rather than from the running
    pipeline: a draft is not deployed, and compiling is both faster and more
    precise about what the author just typed.

    A shared node writes to the library, so the fix reaches every flow using
    it — and that one is live immediately rather than waiting for a publish,
    because the copy is not any single flow's to hold back.
    """
    definition = _read_flow(controller, name)
    ref = _source_ref(definition, node_id)
    if ref:
        changed = await run_in_threadpool(
            controller.store.write_lib_source, ref, source.code
        )
        if changed:
            await controller.reload_lib_users(ref)
    else:
        await run_in_threadpool(
            controller.store.write_node_source, name, node_id, source.code, True
        )
    node_def = next((n for n in definition.nodes if n.id == node_id), None)
    device = (
        node_def.device
        if node_def is not None and node_def.device_policy == "require"
        else None
    )
    error = await run_in_threadpool(
        controller.compile_check, name, node_id, source.code, device
    )
    return NodeStatusPublic(
        id=f"{name}.{node_id}",
        status="error" if error else "active",
        error=error,
    )


@router.post("/{name}/nodes/{node_id}/share", response_model=FlowDetail)
async def share_node(
    name: str,
    node_id: str,
    body: ShareRequest,
    controller: FlowControllerDep,
) -> Any:
    """Move this node's code into the library so other flows can run it too."""
    _read_flow(controller, name)
    if not NAME_PATTERN.match(body.lib_name):
        raise HTTPException(
            status_code=400,
            detail=(
                "Use lowercase letters, digits and underscores, starting with a letter"
            ),
        )
    try:
        await run_in_threadpool(
            controller.store.share_node, name, node_id, body.lib_name
        )
    except LibExists as exc:
        raise HTTPException(status_code=409, detail=str(exc))
    return _detail(controller, _read_flow(controller, name))


@router.post("/{name}/nodes/{node_id}/unshare", response_model=FlowDetail)
async def unshare_node(
    name: str,
    node_id: str,
    controller: FlowControllerDep,
) -> Any:
    """Take a private copy of the shared code back into this flow."""
    _read_flow(controller, name)
    try:
        await run_in_threadpool(controller.store.unshare_node, name, node_id)
    except LibNotFound as exc:
        raise HTTPException(status_code=404, detail=str(exc))
    return _detail(controller, _read_flow(controller, name))


# -----------------------------------------------------------------------------
# Running, stopped, paused
# -----------------------------------------------------------------------------


@router.post("/{name}/start", response_model=FlowDetail)
async def start_flow(
    name: str, controller: FlowControllerDep, user: CurrentUser
) -> Any:
    """Let the engine run this flow again."""
    _read_flow(controller, name)
    await controller.set_enabled(name, True)
    _audit("started", name, user)
    return _detail(controller, _read_flow(controller, name))


@router.post("/{name}/stop", response_model=FlowDetail)
async def stop_flow(name: str, controller: FlowControllerDep, user: CurrentUser) -> Any:
    """Take this flow off the engine: no subscriptions, schedules or webhooks."""
    _read_flow(controller, name)
    await controller.set_enabled(name, False)
    _audit("stopped", name, user)
    return _detail(controller, _read_flow(controller, name))


@router.post("/{name}/pause", response_model=Message)
def pause_flow(name: str, controller: FlowControllerDep) -> Any:
    """Hold the flow's nodes so its messages can be stepped through."""
    _read_flow(controller, name)
    controller.pause_flow(name)
    return Message(message=f"Paused flow '{name}'")


@router.post("/{name}/resume", response_model=Message)
async def resume_flow(name: str, controller: FlowControllerDep) -> Any:
    """Let the flow carry on, running whatever was held back."""
    _read_flow(controller, name)
    await run_in_threadpool(controller.resume_flow, name)
    return Message(message=f"Resumed flow '{name}'")


@router.post("/{name}/step", response_model=Message)
async def step_flow(name: str, controller: FlowControllerDep) -> Any:
    """Run one message a pause is holding back, leaving the flow paused."""
    _read_flow(controller, name)
    node = await run_in_threadpool(controller.step_flow, name)
    if node is None:
        return Message(message=f"Nothing held back in flow '{name}'")
    return Message(message=f"Stepped '{node}'")


# -----------------------------------------------------------------------------
# Validation and execution
# -----------------------------------------------------------------------------


@router.post("/{name}/validate", response_model=ValidationResult)
def validate_flow(name: str, controller: FlowControllerDep) -> Any:
    """Report what would keep this flow from running."""
    _read_flow(controller, name)
    if controller.store.has_draft(name):
        return ValidationResult(issues=controller.preview(name).issues)
    return ValidationResult(issues=controller.flow_issues(name))


@router.post("/{name}/run", response_model=FlowStatePublic)
async def run_flow(
    name: str,
    body: RunRequest,
    request: Request,
    controller: FlowControllerDep,
    user: CurrentUser,
) -> Any:
    """Run every node of a flow once.

    With unpublished changes this runs the draft, so the button matches what is
    on the canvas. Nothing is deployed by running it.

    A *batch* flow is submitted as a run instead, because that is what running
    one means: it is what keeps the parameters, the series and the result, and
    a button that quietly did something else would be a trap.
    """
    definition = _read_flow(controller, name)
    _require_enabled(controller, name)
    if definition.mode == "batch":
        service = getattr(request.app.state, "run_service", None)
        if service is None:
            raise HTTPException(status_code=503, detail="Runs are not available")
        try:
            await run_in_threadpool(
                service.submit,
                name,
                params=body.inputs,
                cause="api",
                actor=user.email,
                draft=controller.store.has_draft(name),
            )
        except RunRejected as exc:
            raise HTTPException(status_code=422, detail=str(exc)) from exc
        return _flow_state(controller, name)

    inputs = {qualify(name, key): value for key, value in body.inputs.items()}
    if controller.store.has_draft(name):
        await run_in_threadpool(controller.run_preview, name, inputs)
    else:
        await run_in_threadpool(controller.run_flow, name, inputs)
    return _flow_state(controller, name)


@router.post("/{name}/nodes/{node_id}/trigger", response_model=FlowStatePublic)
async def trigger_node(
    name: str,
    node_id: str,
    body: TriggerRequest,
    controller: FlowControllerDep,
) -> Any:
    """Feed values into a single node."""
    _require_enabled(controller, name)
    try:
        error = await run_in_threadpool(
            controller.trigger_node, f"{name}.{node_id}", body.values
        )
    except KeyError:
        raise HTTPException(
            status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
        )
    if error:
        # The canvas already has the failure from the bus; the person who
        # clicked gets to read it too, instead of a stack trace in the log.
        raise HTTPException(status_code=400, detail=error)
    return _flow_state(controller, name)


@router.post("/{name}/nodes/{node_id}/cancel", response_model=Message)
def cancel_node(name: str, node_id: str, controller: FlowControllerDep) -> Any:
    """Stop a node that is running right now, by killing the worker running it.

    Idempotent on purpose: by the time a click reaches here the node may well
    have finished, and that is the outcome that was asked for.
    """
    pool = controller.workers
    if pool is not None and pool.cancel(f"{name}.{node_id}"):
        return Message(message=f"Stopped '{node_id}'")
    return Message(message=f"'{node_id}' was not running")


@router.post("/{name}/nodes/{node_id}/acknowledge", response_model=Message)
def acknowledge_node_error(
    name: str, node_id: str, controller: FlowControllerDep
) -> Any:
    """Dismiss what a node last failed with, so the canvas stops marking it.

    A failure outlives the next good run on purpose — otherwise one that fired
    an alert leaves no trace by the time anyone looks. Reading the traceback is
    what says it has been seen.
    """
    try:
        controller.acknowledge_error(f"{name}.{node_id}")
    except KeyError:
        raise HTTPException(
            status_code=404, detail=f"No node named '{node_id}' in flow '{name}'"
        ) from None
    # Another browser holds the same marker and has no reason to refetch, so it
    # would keep showing a failure that is gone until its next snapshot.
    event_bus.publish(
        {
            "type": "node_error_acknowledged",
            "node": f"{name}.{node_id}",
            "ts": time.time(),
        }
    )
    return Message(message=f"Cleared the failure on '{node_id}'")


@router.get("/{name}/state", response_model=FlowStatePublic)
def read_flow_state(name: str, controller: FlowControllerDep) -> Any:
    """The last value seen on every message of this flow."""
    _read_flow(controller, name)
    return _flow_state(controller, name)


@router.get("/{name}/history/{message}", response_model=MessageHistory)
def read_message_history(
    name: str,
    message: str,
    controller: FlowControllerDep,
) -> Any:
    """The recent values of one message, for plotting.

    ``message`` may be given bare or qualified; a message that never carried a
    number or a flag comes back with an empty series.
    """
    _read_flow(controller, name)
    key = qualify(name, message)
    return MessageHistory(
        message=key,
        numeric=as_number(controller.state.get(key)) is not None,
        points=[
            HistoryPoint(ts=ts, value=value)
            for ts, value in controller.state.history(key)
        ],
    )


# -----------------------------------------------------------------------------
# Live updates
# -----------------------------------------------------------------------------


def snapshot_payload(
    controller: FlowController, only: set[str] | None = None
) -> dict[str, Any]:
    """Everything a client needs to catch up, sent the moment it connects.

    Also sent by the tunnel connector, which serves this websocket inline: the
    two have to agree, so they build the message here rather than each their
    own. ``emits`` is what the bus counted while nobody was listening — a
    client that reconnects between two pages would otherwise start from zero.

    ``only`` bounds it to a set of message names, which is what a wall panel
    gets: the values its own dashboards draw, and none of the rest — no node
    status, no logs, no shape of the graph. A panel renders none of that, and
    a screen may be hanging somewhere nobody here can see.
    """
    if only is not None:
        # The same keys, emptied rather than dropped: a screen already hanging
        # runs whatever bundle it was paired with, and the shape of this
        # message is what that bundle reads.
        return {
            "type": "snapshot",
            "values": {k: v for k, v in controller.values().items() if k in only},
            "nodes": [],
            "issues": [],
            "paused": [],
            "logs": [],
            "emits": {},
        }
    return {
        "type": "snapshot",
        "values": controller.values(),
        "nodes": [s.model_dump() for s in controller.node_statuses()],
        "issues": [i.model_dump() for i in controller.issues],
        "paused": controller.paused_flows(),
        "logs": list(event_bus.recent_logs),
        "emits": dict(event_bus.emits),
    }


def panel_scope(token: str, app: Any) -> set[str] | None:
    """The messages this credential is bounded to, or None if it is a person's.

    The socket is the one authenticated surface the route check cannot reach —
    a handshake has no route to judge — so a panel is bounded by what it is
    sent instead of by what it asks for.
    """
    try:
        panel = str(decode_token(token).get("panel") or "")
    except InvalidTokenError:
        return None
    if not panel:
        return None
    store: DashboardStore | None = getattr(app.state, "dashboard_store", None)
    return messages_for(panel, store) if store is not None else set()


def event_for_panel(event: dict[str, Any], only: set[str]) -> bool:
    """Whether a panel's socket should carry this event.

    The same bound as the snapshot above, applied to the stream that follows
    it: a value the panel draws, and nothing else on the bus — save for a
    dashboard being published, which is how a screen hears that the document
    it is drawing, or the set of them it was given, has moved.
    """
    if event.get("type") == "dashboard_changed":
        return True
    return event.get("type") == "message_value" and str(event.get("name") or "") in only


# How many events one frame may carry. A ceiling rather than a target: the
# batch is whatever the bus happens to hold, and a client that has been away
# should not be handed the whole queue in one message.
MAX_FRAME_EVENTS = 64

#: How many message names one socket may ask for bytes on. A screen draws a
#: handful of tiles; this is only here so a client cannot ask for the whole
#: namespace and be served every frame of it.
MAX_MEDIA_NAMES = 32


def wanted_names(frame: str, only: set[str] | None) -> set[str] | None:
    """What this client is asking to be sent bytes for, or None if it said
    something else.

    A client asks by name — the tiles it is drawing — and is answered only for
    the names a panel credential would have been allowed anyway. Bytes are the
    expensive thing on this socket, so nothing is pushed until something says
    it is looking at it.
    """
    try:
        payload = orjson.loads(frame)
    except orjson.JSONDecodeError:
        return None
    if not isinstance(payload, dict) or payload.get("type") != "media":
        return None
    names = payload.get("names")
    if not isinstance(names, list):
        return set()
    asked = {str(name) for name in names[:MAX_MEDIA_NAMES]}
    return asked if only is None else asked & only


def media_frames(
    events: list[dict[str, Any]], wanted: set[str], store: Any
) -> list[bytes]:
    """The bytes behind the media values in this batch, one frame each.

    Referenced-then-fetched costs a round trip per frame, which is what keeps
    the message plane at a glance rather than a view. Pushing the bytes down
    the socket that already carries the event closes that, and only for the
    frames a client said it was drawing.

    Only what the ring holds: the durable store is what a fetch is for, and a
    checkpoint has no business being pushed at anyone. The newest value per
    name wins, so a client that fell behind is not handed a backlog of frames
    it would only draw over.
    """
    if not wanted or store is None or getattr(store, "volatile", None) is None:
        return []

    newest: dict[str, dict[str, Any]] = {}
    for event in events:
        if event.get("type") != "message_value":
            continue
        name = str(event.get("name") or "")
        if name not in wanted or not is_reference(event.get("value")):
            continue
        newest[name] = event

    frames: list[bytes] = []
    for name, event in newest.items():
        value = event["value"]
        digest = str(value.get("digest") or "")
        path = store.volatile.path(digest)
        if path is None:
            continue
        try:
            payload = path.read_bytes()
        except OSError:
            continue
        header = orjson.dumps(
            {
                "type": "media",
                "name": name,
                "digest": digest,
                "media_type": str(value.get("media_type") or ""),
                "ts": event.get("ts"),
            }
        )
        # Length-prefixed, so one frame carries both halves and the reader
        # never has to guess where the JSON stops.
        frames.append(len(header).to_bytes(4, "big") + header + payload)
    return frames


async def _send(websocket: WebSocket, events: list[dict[str, Any]]) -> None:
    """One event, or a batch of them under `events`.

    Serialised once with orjson rather than per client with the stdlib, which
    is what `send_json` does.
    """
    payload = events[0] if len(events) == 1 else {"type": "batch", "events": events}
    await websocket.send_text(orjson.dumps(payload).decode())


@ws_router.websocket("/ws")
async def flow_events(websocket: WebSocket, token: str = "") -> None:
    """Stream values, node status and execution events as they happen.

    The token goes in the query string because browsers cannot set headers on
    a websocket handshake.
    """

    # On a thread: authenticating is a database round trip, and the snapshot
    # below reads the whole of state. Neither belongs on the event loop, which
    # every other socket and every request is sharing.
    def _authenticate() -> Any:
        with Session(engine) as session:
            return user_from_token(session, token)

    user = await run_in_threadpool(_authenticate)
    if user is None:
        await websocket.close(code=1008)
        return
    only = panel_scope(token, websocket.app)
    # Nothing until a client says it is drawing something: bytes are what this
    # socket cannot afford to send speculatively.
    wanted: set[str] = set()

    await websocket.accept()

    controller: FlowController | None = getattr(
        websocket.app.state, "flow_controller", None
    )
    store = getattr(websocket.app.state, "artifact_store", None)

    async def send_snapshot() -> None:
        if controller is not None:
            payload = await run_in_threadpool(snapshot_payload, controller, only)
            await websocket.send_text(orjson.dumps(payload).decode())

    await send_snapshot()

    async with event_bus.subscribe() as queue:
        receiver = asyncio.create_task(websocket.receive_text())
        try:
            while True:
                sender = asyncio.create_task(queue.get())
                done, _ = await asyncio.wait(
                    {sender, receiver}, return_when=asyncio.FIRST_COMPLETED
                )
                if receiver in done:
                    if sender not in done:
                        # Nobody will ever read this one: the next turn makes
                        # its own. Left behind it stays parked on the queue's
                        # waiter list, takes an event the live reader wanted,
                        # and is reported as a task destroyed while pending.
                        sender.cancel()
                    # A frame from the client. Only a disconnect ends the
                    # stream — a keepalive, or anything else it decides to
                    # say, used to be read as the client going away and cost
                    # it every live update from then on.
                    failure = receiver.exception()
                    if failure is not None:
                        # The client is gone, and `receive_text` raises for
                        # good from here on. Reading it again would spin this
                        # loop at full tilt against a dead socket.
                        raise failure
                    asked = wanted_names(receiver.result(), only)
                    if asked is not None:
                        wanted = asked
                    receiver = asyncio.create_task(websocket.receive_text())
                    if sender not in done:
                        continue
                # `sender` is done from here: `FIRST_COMPLETED` returns with at
                # least one of the two, so `receiver` missing means this one is.

                # Everything the bus has right now, not just the one event
                # that woke this: a cascade puts a dozen in at once, and one
                # frame carrying them costs one wakeup rather than a dozen.
                batch = [sender.result()]
                while len(batch) < MAX_FRAME_EVENTS:
                    try:
                        batch.append(queue.get_nowait())
                    except asyncio.QueueEmpty:
                        break

                out: list[dict[str, Any]] = []
                for event in batch:
                    if only is not None and event.get("type") == "dashboard_changed":
                        # The scope was resolved once, at the handshake. A
                        # panel pointed at another dashboard would otherwise
                        # fetch the new document and then draw tiles nothing
                        # ever updates. ``or set()`` because a panel that was
                        # deleted resolves to None, the same as a person's
                        # token — and that would widen this socket to
                        # everything on the bus.
                        only = panel_scope(token, websocket.app) or set()
                        wanted &= only
                        if out:
                            await _send(websocket, out)
                            out = []
                        await send_snapshot()
                    if only is not None and not event_for_panel(event, only):
                        continue
                    out.append(event)
                if out:
                    # The bytes first: a tile that has the frame when the
                    # value lands draws it in one pass rather than two.
                    for frame in media_frames(out, wanted, store):
                        await websocket.send_bytes(frame)
                    await _send(websocket, out)
        except (WebSocketDisconnect, RuntimeError):
            # A peer that goes away mid-send takes the RuntimeError route
            # ("websocket.send after websocket.close") rather than the clean
            # disconnect. Either way the socket is gone and the loop is over.
            pass
        finally:
            receiver.cancel()