app/backend/fluksio/flow/namespace.py

"""The names the engine keeps live state and queued work under, and how a
rename of a flow key maps them.

A flow's messages and nodes are named ``<key>.<name>``, and the engine's own
bookkeeping on them sits in state beside the values, under a reserved prefix
and the names it is about: ``__version__:heat.temp``,
``__last_seen__:heat.valve:hall.temp``. Renaming a flow key moves all of it to
the new name, so every family is registered here, beside the builders that
take their prefix from it, and a key of a family that is not is refused rather
than left behind.

Documents name messages too: a flow its ports, inputs and outputs, a dashboard
its bindings. Where they do is registered beside the paths into them
(`FLOW_FIELDS`, `dashboards.DASHBOARD_BINDINGS`), and `rebind` rewrites them.

Nothing here does I/O. It maps names; moving the keys and items is the state
and queue backends' work, and writing documents the rename's.
"""

from __future__ import annotations

from collections.abc import Callable, Iterable
from dataclasses import dataclass, replace
from typing import TYPE_CHECKING, Any

import orjson

from fluksio.flow.messages import flow_of

if TYPE_CHECKING:
    from fluksio.flow.queue import WorkItem

#: The message names in state, as a set (`RedisState` only).
NAMES = "__names__"
#: A message's recorded points (`RedisState`; memory keeps them apart).
HISTORY = "__history__"
#: How often a message was published.
VERSION = "__version__"
#: When a message was last published.
TIMESTAMP = "__ts__"
#: The version of a message a node last ran on.
LAST_SEEN = "__last_seen__"
#: When a rate-limited input last woke a node.
DELIVERED = "__in_ts__"
#: The value a rate-limited output kept back.
HELD = "__held__"
#: When a node's rate-limit window is due to be let through.
FLUSH = "__flush__"
#: What an engine-shipped node type remembers (`Node.remember`).
NODE_STATE = "__node__"
#: Not state: the work queue's stream from before the queue had a root of its
#: own. Startup deletes it unless it holds work nobody delivered
#: (`RedisWorkQueue.drop_legacy`). Kept, it names no flow's state, and the
#: queue counts that work as queued instead.
LEGACY_QUEUE = "__queue__"

#: A qualified message or node name, ``<flow key>.<name>``.
NAME = "name"
#: A node's own key, user data that may contain colons.
PRIVATE = "private"
#: A flow key.
FLOW = "flow"
#: A map keyed by qualified message names.
OUTPUTS = "outputs"

#: Every bookkeeping family in live state, by prefix, with the parts that
#: follow it. A key is ``<prefix>:<part>:<part>``. Names never contain a colon,
#: so the key splits at its first ones and the last part takes the rest, which
#: is how a private key keeps its own.
FAMILIES: dict[str, tuple[str, ...]] = {
    NAMES: (),
    HISTORY: (NAME,),
    VERSION: (NAME,),
    TIMESTAMP: (NAME,),
    LAST_SEEN: (NAME, NAME),
    DELIVERED: (NAME, NAME),
    HELD: (NAME,),
    FLUSH: (NAME,),
    NODE_STATE: (NAME, PRIVATE),
    LEGACY_QUEUE: (),
}

#: What each `WorkItem` field holds that a rename changes, or None. A guard is
#: the node's own key, and the state it guards moves with the node.
ITEM_FIELDS: dict[str, str | None] = {
    "kind": None,
    "node": NAME,
    "flow": FLOW,
    "outputs": OUTPUTS,
    "cause": None,
    "not_before": None,
    "guard_key": None,
    "guard_value": None,
    "entry_id": None,
    "deliveries": None,
    "enqueued_at": None,
    "run_id": None,
    "parent": None,
}

#: Where a flow document names messages, as paths into its JSON (`rebind`).
#: A bare name is the flow's own and follows its key by itself, so a rename
#: moves only the qualified ones: those the flow names in another's
#: namespace, or explicitly in its own.
FLOW_FIELDS: tuple[tuple[str, ...], ...] = (
    ("nodes", "*", "requires", "*", "name"),
    ("nodes", "*", "provides", "*", "name"),
    ("inputs", "*", "spec", "name"),
    ("outputs", "*"),
)


def rebind(
    document: Any, paths: Iterable[tuple[str, ...]], rename: Callable[[str], str]
) -> int:
    """Pass every string *paths* lead to in *document* through *rename*, in
    place, and count those that changed.

    A path is a key per level, or ``*`` for every item of a list or value of
    an object. One that leads nowhere, or to anything but a string, is passed
    over, so a document missing a field or holding an older shape of it is
    read as far as it goes.
    """
    return sum(_rebind(document, path, rename) for path in paths)


def names(document: Any, paths: Iterable[tuple[str, ...]]) -> list[str]:
    """Every string *paths* lead to in *document* (`rebind`)."""
    found: list[str] = []

    def keep(name: str) -> str:
        found.append(name)
        return name

    rebind(document, paths, keep)
    return found


def _rebind(node: Any, path: tuple[str, ...], rename: Callable[[str], str]) -> int:
    head, rest = path[0], path[1:]
    keys: Iterable[Any] = ()
    if head == "*" and isinstance(node, list):
        keys = range(len(node))
    elif head == "*" and isinstance(node, dict):
        keys = list(node)
    elif isinstance(node, dict) and head in node:
        keys = (head,)
    changed = 0
    for key in keys:
        if rest:
            changed += _rebind(node[key], rest, rename)
        elif isinstance(node[key], str) and (moved := rename(node[key])) != node[key]:
            node[key] = moved
            changed += 1
    return changed


def _parts(key: str) -> tuple[str, list[tuple[str, str]]]:
    """A state key's family prefix and its parts, each with what it holds.

    A key without a prefix is a message value, and its family is ``""``.
    """
    if not key.startswith("__"):
        return "", [(key, NAME)]
    family, sep, rest = key.partition(":")
    shape = FAMILIES.get(family)
    parts = rest.split(":", len(shape) - 1) if shape is not None and sep else []
    if shape is None or len(parts) != len(shape):
        raise ValueError(f"State key '{key}' is in no registered family")
    return family, list(zip(parts, shape, strict=True))


def touches(flow: str, key: str) -> bool:
    """Whether the state key *key* names anything of the flow key *flow*."""
    _, parts = _parts(key)
    return any(kind == NAME and flow_of(part) == flow for part, kind in parts)


@dataclass(frozen=True)
class Rename:
    """The flow key *old* becoming *new*, applied to the names that embed it."""

    old: str
    new: str

    def name(self, qualified: str) -> str:
        """A qualified message or node name in the new namespace.

        Only the part before the first dot is the flow, so renaming ``heat``
        leaves ``heating.temp`` alone. A bare key is its own namespace.
        """
        if flow_of(qualified) != self.old:
            return qualified
        return self.new + qualified[len(self.old) :]

    def flow(self, key: str) -> str:
        return self.new if key == self.old else key

    def binding(self, name: str) -> str:
        """A message name as a document holds it. A bare one is local to the
        flow the document is, so only a qualified one moves."""
        return self.name(name) if "." in name else name

    def state_key(self, key: str) -> str | None:
        """Where the state key *key* moves to, or None if it stays.

        :raises ValueError: For a bookkeeping key of an unregistered family.
        """
        family, parts = _parts(key)
        mapped = [self.name(part) if kind == NAME else part for part, kind in parts]
        renamed = ":".join([family, *mapped]) if family else mapped[0]
        return renamed if renamed != key else None

    def item(self, item: WorkItem) -> WorkItem | None:
        """*item* with its names in the new namespace, or None if it has none.

        A ``run`` item names its flow by id, which a rename leaves as it is.
        """
        if item.kind == "run":
            return None
        changes = {}
        for field, holds in ITEM_FIELDS.items():
            if holds is not None:
                value = getattr(item, field)
                mapped = self._mapped(holds, value)
                if mapped != value:
                    changes[field] = mapped
        return replace(item, **changes) if changes else None

    def fields(self, fields: dict[str, str]) -> dict[str, str] | None:
        """A serialized item (`WorkItem.to_fields`) in the new namespace, or
        None if it names nothing renamed. Fields a rename does not touch keep
        their exact text, so an unchanged item serializes as it did."""
        if fields.get("kind") == "run":
            return None
        renamed = dict(fields)
        for field, holds in ITEM_FIELDS.items():
            if holds is None or field not in fields:
                continue
            if holds == OUTPUTS:
                outputs = orjson.loads(fields[field] or "{}")
                mapped = self._mapped(holds, outputs)
                if mapped != outputs:
                    renamed[field] = orjson.dumps(mapped).decode()
            else:
                renamed[field] = self._mapped(holds, fields[field])
        return renamed if renamed != fields else None

    def dedup_key(self, key: str) -> str:
        """An alert's ``<event type>:<node or flow>`` (`alerts.dedup_key`)."""
        event, sep, about = key.partition(":")
        return f"{event}{sep}{self.name(about)}"

    def _mapped(self, holds: str, value: Any) -> Any:
        if holds == NAME:
            return self.name(value)
        if holds == FLOW:
            return self.flow(value)
        return {self.name(name): one for name, one in value.items()}