"""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 ``.``, 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, ``.``. 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 ``::``. 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 ``:`` (`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()}