From 847abd67d6153cb2a91fa5b8e2ea3f21ba023f73 Mon Sep 17 00:00:00 2001 From: stroblme Date: Tue, 6 Oct 2026 17:19:48 +0200 Subject: [PATCH] Give connectors a store that survives restarts, and a way to clear it `ConnectorNode.store` keeps short strings per connector package in the engine's database (`connector_state`, migration f8c3e5a91d27), so the nodes of one package share them and `fluksio serve` needs nothing new: get, set and delete, at most 64 kB a value and 256 keys a package. A superuser lists the keys (never the values) and clears them through `/connectors/{node_type}/state`, or under Stored state in a connector node's panel. Documented in the connector contract. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_013GAeg4Jo6xjZzjwy9QDWMQ --- .../versions/f8c3e5a91d27_connector_state.py | 38 +++++ backend/fluksio/api/main.py | 2 + backend/fluksio/api/routes/connectors.py | 79 ++++++++++ backend/fluksio/flow/connector.py | 21 ++- backend/fluksio/flow/connector_store.py | 139 ++++++++++++++++++ backend/fluksio/models.py | 22 +++ backend/tests/api/routes/test_connectors.py | 74 ++++++++++ backend/tests/test_connector_store.py | 65 ++++++++ docs/code/api.md | 5 +- docs/code/connectors.md | 1 + docs/reference/connector-contract.md | 42 +++++- frontend/src/client/schemas.gen.ts | 44 ++++++ frontend/src/client/sdk.gen.ts | 69 ++++++++- frontend/src/client/types.gen.ts | 31 ++++ frontend/src/components/Flow/NodePanel.tsx | 82 +++++++++++ 15 files changed, 710 insertions(+), 4 deletions(-) create mode 100644 backend/fluksio/alembic/versions/f8c3e5a91d27_connector_state.py create mode 100644 backend/fluksio/api/routes/connectors.py create mode 100644 backend/fluksio/flow/connector_store.py create mode 100644 backend/tests/api/routes/test_connectors.py create mode 100644 backend/tests/test_connector_store.py diff --git a/backend/fluksio/alembic/versions/f8c3e5a91d27_connector_state.py b/backend/fluksio/alembic/versions/f8c3e5a91d27_connector_state.py new file mode 100644 index 0000000..3bf4ca9 --- /dev/null +++ b/backend/fluksio/alembic/versions/f8c3e5a91d27_connector_state.py @@ -0,0 +1,38 @@ +"""connector_state + +A place for a connector to keep what has to outlive a restart, such as the +certificate a Hue bridge presented on first contact. Keyed by the connector's +package, so the nodes of one package share it. + +Revision ID: f8c3e5a91d27 +Revises: d5a3c8e1f246 +Create Date: 2026-10-06 + +""" + +import sqlalchemy as sa +import sqlmodel.sql.sqltypes +from alembic import op + +# revision identifiers, used by Alembic. +revision = "f8c3e5a91d27" +down_revision = "d5a3c8e1f246" +branch_labels = None +depends_on = None + + +def upgrade(): + op.create_table( + "connector_state", + sa.Column( + "connector", sqlmodel.sql.sqltypes.AutoString(length=100), nullable=False + ), + sa.Column("key", sqlmodel.sql.sqltypes.AutoString(length=200), nullable=False), + sa.Column("value", sqlmodel.sql.sqltypes.AutoString(), nullable=False), + sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False), + sa.PrimaryKeyConstraint("connector", "key"), + ) + + +def downgrade(): + op.drop_table("connector_state") diff --git a/backend/fluksio/api/main.py b/backend/fluksio/api/main.py index 0d4f8b2..36c2130 100644 --- a/backend/fluksio/api/main.py +++ b/backend/fluksio/api/main.py @@ -5,6 +5,7 @@ from fluksio.api.routes import ( alerts, artifacts, cloud, + connectors, dashboards, flavors, flows, @@ -31,6 +32,7 @@ api_router.include_router(flavors.router) api_router.include_router(flows.router) api_router.include_router(flows.ws_router) api_router.include_router(secrets.router) +api_router.include_router(connectors.router) api_router.include_router(alerts.router) api_router.include_router(dashboards.router) api_router.include_router(panels.router) diff --git a/backend/fluksio/api/routes/connectors.py b/backend/fluksio/api/routes/connectors.py new file mode 100644 index 0000000..3a5c3d2 --- /dev/null +++ b/backend/fluksio/api/routes/connectors.py @@ -0,0 +1,79 @@ +"""What a connector keeps across restarts, for a superuser to see and reset. + +A connector writes these itself (``ConnectorNode.store``). This is the way back +when one of them is wrong: a Hue bridge replaced, so the certificate pinned for +it no longer matches. Values are never handed out, since a connector may keep a +token here; the keys say enough to decide what to clear. + +A connector is named by any node type it installs, as the canvas knows it, and +its keys are shared by every node type of the same package. +""" + +from datetime import datetime +from typing import Any + +from fastapi import APIRouter, Depends, HTTPException +from pydantic import BaseModel + +from fluksio.api.deps import get_current_active_superuser +from fluksio.flow.connector import ConnectorNode +from fluksio.flow.connector_store import KEY, ConnectorStore, namespace_of +from fluksio.flow.controller import NODE_TYPES +from fluksio.models import Message + +router = APIRouter( + prefix="/connectors", + tags=["connectors"], + dependencies=[Depends(get_current_active_superuser)], +) + + +class StoredKey(BaseModel): + key: str + #: Bytes, so a superuser can tell a fingerprint from a blob. + size: int + updated_at: datetime + + +class ConnectorStatePublic(BaseModel): + #: The package the keys belong to. + connector: str + data: list[StoredKey] + count: int + + +def _store(node_type: str) -> ConnectorStore: + """The store of the connector that installed *node_type*, or a 404.""" + known = NODE_TYPES.get(node_type) + cls = known.cls if known is not None else None + if not (isinstance(cls, type) and issubclass(cls, ConnectorNode)): + raise HTTPException(status_code=404, detail=f"No connector named '{node_type}'") + return ConnectorStore(namespace_of(cls)) + + +@router.get("/{node_type}/state", response_model=ConnectorStatePublic) +def read_connector_state(node_type: str) -> Any: + """The keys this connector keeps, without their values.""" + store = _store(node_type) + keys = [ + StoredKey(key=row.key, size=len(row.value.encode()), updated_at=row.updated_at) + for row in store.entries() + ] + return ConnectorStatePublic(connector=store.connector, data=keys, count=len(keys)) + + +@router.delete("/{node_type}/state", response_model=Message) +def clear_connector_state(node_type: str) -> Any: + """Forget everything this connector keeps. It starts over on next use.""" + store = _store(node_type) + cleared = store.clear() + return Message(message=f"Cleared {cleared} key(s) of {store.connector}") + + +@router.delete("/{node_type}/state/{key}", response_model=Message) +def delete_connector_state(node_type: str, key: str) -> Any: + """Forget one key, such as the certificate pinned for one bridge.""" + store = _store(node_type) + if not KEY.fullmatch(key) or not store.delete(key): + raise HTTPException(status_code=404, detail=f"No key '{key}' stored") + return Message(message=f"Deleted '{key}' of {store.connector}") diff --git a/backend/fluksio/flow/connector.py b/backend/fluksio/flow/connector.py index de7a378..0dec6f5 100644 --- a/backend/fluksio/flow/connector.py +++ b/backend/fluksio/flow/connector.py @@ -20,7 +20,10 @@ What a connector gets from the base class: reading it; * :meth:`ConnectorNode.save_artifact`, for a device whose readings are bytes — a camera frame, a recorded clip — which travel as a reference on a media - port rather than as the message itself. + port rather than as the message itself; +* :attr:`ConnectorNode.store`, for the few short strings that have to outlive + a restart — a certificate pinned on first contact, a pairing token — shared + by every node of the connector's package. The message schemas and the parameter model are the rest of the contract, and they are the same ones the built-in nodes use. See ``docs/connectors/`` for the @@ -35,6 +38,7 @@ from typing import TYPE_CHECKING, Any, ClassVar from pydantic import BaseModel, Field +from fluksio.flow.connector_store import ConnectorStore, namespace_of from fluksio.flow.nodes import Node if TYPE_CHECKING: @@ -92,6 +96,7 @@ class ConnectorNode(Node): "_stop_event", "_last_published", "_artifacts", + "_store", "_down", ) @@ -102,6 +107,7 @@ class ConnectorNode(Node): self._stop_event: asyncio.Event | None = None self._last_published: dict[str, Any] = {} self._artifacts: ArtifactStore | None = None + self._store: ConnectorStore | None = None self._down = False def _dispatch(self, params: dict[str, Any], **ports: Any) -> dict[str, Any] | None: @@ -166,6 +172,19 @@ class ConnectorNode(Node): [data], name=name, media_type=media_type, volatile=volatile ) + @property + def store(self) -> ConnectorStore: + """Short strings this connector keeps across restarts. + + ``get(key)``, ``set(key, value)`` and ``delete(key)``, namespaced by the + connector's top-level package, so every node of the package sees the + same keys. Blocking: call it through ``asyncio.to_thread`` from the + event loop. Available before :meth:`start`, unlike the artifact store. + """ + if self._store is None: + self._store = ConnectorStore(namespace_of(type(self))) + return self._store + async def start(self, app: FastAPI | None = None) -> None: self._artifacts = getattr(app.state, "artifact_store", None) if app else None if self.config.poll_interval > 0 and self._stop_event is None: diff --git a/backend/fluksio/flow/connector_store.py b/backend/fluksio/flow/connector_store.py new file mode 100644 index 0000000..ab85355 --- /dev/null +++ b/backend/fluksio/flow/connector_store.py @@ -0,0 +1,139 @@ +"""What a connector keeps across restarts. + +A connector holds connections and handles, and those are meant to go when the +process does. A few things are not: the certificate a device presented on first +contact, a token it paired with. :attr:`ConnectorNode.store` is where those go — +a handful of short strings per connector package, in the engine's own database, +so they survive a restart and need nothing ``fluksio serve`` does not have. + +Keyed by the connector's top-level package rather than by node: the nodes of +one Hue bridge share the certificate pinned for it, and every node type a +package installs shares the same keys. A superuser lists and clears them +through ``/connectors/{node_type}/state``. +""" + +from __future__ import annotations + +import re +from datetime import UTC, datetime + +from sqlalchemy import delete, func +from sqlmodel import Session, col, select + +from fluksio.models import ConnectorState + +#: A value is a short string — a fingerprint, a PEM, a token. Bytes worth +#: keeping belong in the artifact store. +MAX_VALUE_BYTES = 64 * 1024 +#: Keys per connector package. Past this a connector is keeping something per +#: reading rather than per device. +MAX_KEYS = 256 +#: What a top-level package can be called. +CONNECTOR_NAME = re.compile(r"[A-Za-z_][A-Za-z0-9_]{0,99}") +#: One URL path segment, so the reset route can name every key. +KEY = re.compile(r"[^/\s]{1,200}") + + +def namespace_of(cls: type) -> str: + """The package a connector class keeps its state under.""" + return cls.__module__.partition(".")[0] + + +def _checked(key: str) -> str: + if not KEY.fullmatch(key): + raise ValueError( + f"'{key}' is not a valid key: 1 to 200 characters, no slash or whitespace" + ) + return key + + +class ConnectorStore: + """Short strings one connector package keeps across restarts. + + Every call is a database round trip, so from the event loop it goes + through ``asyncio.to_thread`` like any other blocking I/O. + """ + + def __init__(self, connector: str) -> None: + if not CONNECTOR_NAME.fullmatch(connector): + raise ValueError(f"'{connector}' is not a connector package name") + self.connector = connector + + def get(self, key: str) -> str | None: + """The value stored under *key*, or ``None``.""" + # Imported here: a connector imports this module, and its tests have + # no business opening the engine's database by importing it. + from fluksio.core.db import engine + + with Session(engine) as session: + row = session.get(ConnectorState, (self.connector, _checked(key))) + return None if row is None else row.value + + def set(self, key: str, value: str) -> None: + """Store *value* under *key*, replacing what was there.""" + from fluksio.core.db import engine, writing + + _checked(key) + if len(value.encode()) > MAX_VALUE_BYTES: + raise ValueError( + f"'{key}' is too large to store: {MAX_VALUE_BYTES} bytes at most" + ) + with Session(engine) as session, writing(session): + row = session.get(ConnectorState, (self.connector, key)) + if row is None: + count = session.exec( + select(func.count()) + .select_from(ConnectorState) + .where(ConnectorState.connector == self.connector) + ).one() + if count >= MAX_KEYS: + raise ValueError( + f"{self.connector} already stores {MAX_KEYS} keys, the most " + "a connector may keep" + ) + row = ConnectorState(connector=self.connector, key=key, value=value) + else: + row.value = value + row.updated_at = datetime.now(UTC) + session.add(row) + session.commit() + + def delete(self, key: str) -> bool: + """Forget *key*. Whether there was anything to forget.""" + from fluksio.core.db import engine + + with Session(engine) as session: + result = session.exec( + delete(ConnectorState).where( + col(ConnectorState.connector) == self.connector, + col(ConnectorState.key) == _checked(key), + ) + ) + session.commit() + return bool(result.rowcount) + + def entries(self) -> list[ConnectorState]: + """Every key this package keeps, in key order.""" + from fluksio.core.db import engine + + with Session(engine) as session: + return list( + session.exec( + select(ConnectorState) + .where(ConnectorState.connector == self.connector) + .order_by(col(ConnectorState.key)) + ).all() + ) + + def clear(self) -> int: + """Forget every key. How many there were.""" + from fluksio.core.db import engine + + with Session(engine) as session: + result = session.exec( + delete(ConnectorState).where( + col(ConnectorState.connector) == self.connector + ) + ) + session.commit() + return int(result.rowcount) diff --git a/backend/fluksio/models.py b/backend/fluksio/models.py index 8fc118a..f924275 100644 --- a/backend/fluksio/models.py +++ b/backend/fluksio/models.py @@ -544,3 +544,25 @@ class FlavorPublic(FlavorBase): class FlavorsPublic(SQLModel): data: list[FlavorPublic] count: int + + +# ----------------------------------------------------------------------------- +# Connector state +# +# What a connector has to remember across restarts and has no other home for: +# a certificate it pinned on first contact, a pairing token a device handed it. +# Small values by design — `fluksio.flow.connector_store` caps them. +# ----------------------------------------------------------------------------- + + +class ConnectorState(SQLModel, table=True): + """One value a connector package stored under a key of its own.""" + + __tablename__ = "connector_state" + + #: The connector's top-level package, so every node type it installs and + #: every node of those shares one set of keys. + connector: str = Field(primary_key=True, max_length=100) + key: str = Field(primary_key=True, max_length=200) + value: str + updated_at: datetime = Field(default_factory=get_datetime_utc, sa_type=UTCDateTime) diff --git a/backend/tests/api/routes/test_connectors.py b/backend/tests/api/routes/test_connectors.py new file mode 100644 index 0000000..9b23b1e --- /dev/null +++ b/backend/tests/api/routes/test_connectors.py @@ -0,0 +1,74 @@ +"""Resetting what a connector keeps: a superuser's, named by node type.""" + +import pytest +from fastapi.testclient import TestClient + +from fluksio.core.config import settings +from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode +from fluksio.flow.controller import NODE_TYPES, NodeType + +URL = f"{settings.API_V1_STR}/connectors" + + +class Bridge(ConnectorNode): + contract = CONTRACT_VERSION + + +@pytest.fixture +def bridge(monkeypatch): + monkeypatch.setitem(NODE_TYPES, "test_bridge", NodeType("Bridge", "", Bridge)) + store = Bridge(name="b").store + store.set("cert:10.0.0.2", "PEM") + store.set("cert:10.0.0.3", "PEM") + yield store + store.clear() + + +def test_a_superuser_lists_and_clears_the_keys( + client: TestClient, superuser_token_headers: dict[str, str], bridge +) -> None: + listed = client.get(f"{URL}/test_bridge/state", headers=superuser_token_headers) + assert listed.status_code == 200 + body = listed.json() + assert body["connector"] == "tests" + assert [row["key"] for row in body["data"]] == ["cert:10.0.0.2", "cert:10.0.0.3"] + assert "value" not in body["data"][0] + + one = client.delete( + f"{URL}/test_bridge/state/cert:10.0.0.2", headers=superuser_token_headers + ) + assert one.status_code == 200 + assert bridge.get("cert:10.0.0.2") is None + + rest = client.delete(f"{URL}/test_bridge/state", headers=superuser_token_headers) + assert rest.status_code == 200 + assert bridge.entries() == [] + + +def test_anyone_else_is_refused( + client: TestClient, normal_user_token_headers: dict[str, str], bridge +) -> None: + for method in ("GET", "DELETE"): + response = client.request( + method, f"{URL}/test_bridge/state", headers=normal_user_token_headers + ) + assert response.status_code == 403 + assert bridge.get("cert:10.0.0.2") == "PEM" + + +@pytest.mark.parametrize("path", ["python/state", "nothing/state", "..%2F../state"]) +def test_only_a_connector_has_state( + client: TestClient, superuser_token_headers: dict[str, str], path: str +) -> None: + response = client.get(f"{URL}/{path}", headers=superuser_token_headers) + assert response.status_code == 404 + + +@pytest.mark.usefixtures("bridge") +def test_a_key_that_is_not_there_is_a_404( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + response = client.delete( + f"{URL}/test_bridge/state/cert:nowhere", headers=superuser_token_headers + ) + assert response.status_code == 404 diff --git a/backend/tests/test_connector_store.py b/backend/tests/test_connector_store.py new file mode 100644 index 0000000..5b27228 --- /dev/null +++ b/backend/tests/test_connector_store.py @@ -0,0 +1,65 @@ +"""What a connector keeps across restarts: per package, small, and durable.""" + +import pytest + +from fluksio.flow import connector_store +from fluksio.flow.connector import CONTRACT_VERSION, ConnectorNode +from fluksio.flow.connector_store import MAX_VALUE_BYTES, ConnectorStore + + +class Bridge(ConnectorNode): + contract = CONTRACT_VERSION + + +@pytest.fixture +def store(): + store = ConnectorStore("store_test") + yield store + store.clear() + + +def test_the_nodes_of_one_package_share_its_keys(): + first, second = Bridge(name="a"), Bridge(name="b") + try: + first.store.set("cert:10.0.0.2", "PEM") + assert first.store.connector == "tests" + assert second.store.get("cert:10.0.0.2") == "PEM" + assert ConnectorStore("other").get("cert:10.0.0.2") is None + finally: + first.store.clear() + + +def test_a_value_survives_a_fresh_store(store): + store.set("token", "one") + store.set("token", "two") + + again = ConnectorStore("store_test") + assert again.get("token") == "two" + assert again.delete("token") is True + assert again.delete("token") is False + assert store.get("token") is None + + +def test_a_value_or_a_key_count_past_the_limit_is_refused(store, monkeypatch): + with pytest.raises(ValueError, match="too large"): + store.set("big", "x" * (MAX_VALUE_BYTES + 1)) + + monkeypatch.setattr(connector_store, "MAX_KEYS", 2) + store.set("one", "1") + store.set("two", "2") + with pytest.raises(ValueError, match="2 keys"): + store.set("three", "3") + # Replacing a key that is already there is not a new one. + store.set("two", "22") + assert [row.key for row in store.entries()] == ["one", "two"] + + +@pytest.mark.parametrize("key", ["", "a/b", "a b", "x" * 201]) +def test_a_key_the_reset_route_could_not_name_is_refused(store, key): + with pytest.raises(ValueError, match="not a valid key"): + store.set(key, "v") + + +def test_a_package_name_is_a_name(): + with pytest.raises(ValueError): + ConnectorStore("../flows") diff --git a/docs/code/api.md b/docs/code/api.md index 8214aaa..9b4e754 100644 --- a/docs/code/api.md +++ b/docs/code/api.md @@ -204,7 +204,10 @@ curl -X POST $FLUKSIO/alerts/test/my-phone -H "Authorization: Bearer $TOKEN" ``` Secrets are write-only over the API: you can list names and set values, never -read one back. Secrets and `POST /modules/apply` are a superuser's. +read one back. Secrets and `POST /modules/apply` are a superuser's, and so is +what a connector keeps across restarts: `GET` or `DELETE +/connectors/{node_type}/state`, or `DELETE /connectors/{node_type}/state/{key}` +for one key (see [the connector contract](../reference/connector-contract.md#keeping-state-across-restarts)). ## Observability diff --git a/docs/code/connectors.md b/docs/code/connectors.md index e0a138c..d6b66f8 100644 --- a/docs/code/connectors.md +++ b/docs/code/connectors.md @@ -143,3 +143,4 @@ raised exception shows there with its traceback and marks the node down. - `poll()` returns `None` rather than raising when there is simply nothing new. - `write()` returns quickly; the thread it runs on is the scheduler's. - Health reported when the connection comes and goes. +- Anything that has to outlive a restart in `self.store`, keyed by device. diff --git a/docs/reference/connector-contract.md b/docs/reference/connector-contract.md index 9f897b8..d46b2de 100644 --- a/docs/reference/connector-contract.md +++ b/docs/reference/connector-contract.md @@ -201,6 +201,45 @@ a remote panel wants a frame every second or two; higher rates than that are for the local network, and full-rate video still belongs on the camera's own stream. +## Keeping state across restarts + +A connection, a session or a device handle lives in the connector's own +attributes and goes with the process. The few things that must not, such as +the certificate a device presented on first contact or a token it paired with, +go in `self.store`: + +```python +pem = await asyncio.to_thread(self.store.get, f"cert:{self.config.host}") +await asyncio.to_thread(self.store.set, f"cert:{self.config.host}", pem) +await asyncio.to_thread(self.store.delete, f"cert:{self.config.host}") +``` + +- Keys and values are strings; `get` returns `None` for a key that is not + there, and `delete` returns whether there was one. +- The store belongs to the connector's **top-level package**, not to a node: + every node of the package sees the same keys, so two nodes on one bridge + share what was stored for it. Put the device in the key. +- It lives in the engine's database, so it survives a restart and needs + nothing `fluksio serve` does not have. It is available before `start`. +- Limits: a key is 1 to 200 characters without `/` or whitespace, a value at + most 64 kB, and a package at most 256 keys. Past any of them, `set` raises + `ValueError`. Bytes worth keeping belong in `save_artifact`. +- Every call is a database round trip, so on the event loop it goes through + `asyncio.to_thread`. + +A superuser can see the keys (never the values) and clear them, all of them or +one at a time, under *Stored state* in a connector node's settings, or through +the API, which names the connector by any node type it installs: + +```sh +curl -s $FLUKSIO/connectors/hue/state -H "Authorization: Bearer $TOKEN" +curl -X DELETE $FLUKSIO/connectors/hue/state/cert:10.0.0.2 -H "Authorization: Bearer $TOKEN" +curl -X DELETE $FLUKSIO/connectors/hue/state -H "Authorization: Bearer $TOKEN" +``` + +So a connector has to cope with a key disappearing: clearing is how somebody +tells it to start over. + ## Lifecycle ```python @@ -239,7 +278,8 @@ load is reported separately and never runs. - **Hold state that belongs in the flow.** Logic nodes are stateless and run in parallel; a connector may hold a connection, a session, or a device handle, - but a value another node needs belongs in a message. + and keep what has to outlive a restart in `self.store`, but a value another + node needs belongs in a message. - **Block the event loop.** `poll`, `start` and `stop` are async and run on the loop. Use `asyncio.to_thread` for blocking I/O. - **Read credentials from the environment.** They come through parameters, so diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index 98a1435..0a167f4 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -565,6 +565,29 @@ export const ChatItemSchema = { title: 'ChatItem' } as const; +export const ConnectorStatePublicSchema = { + properties: { + connector: { + type: 'string', + title: 'Connector' + }, + data: { + items: { + '$ref': '#/components/schemas/StoredKey' + }, + type: 'array', + title: 'Data' + }, + count: { + type: 'integer', + title: 'Count' + } + }, + type: 'object', + required: ['connector', 'data', 'count'], + title: 'ConnectorStatePublic' +} as const; + export const ControlNodeSchema = { properties: { id: { @@ -3669,6 +3692,27 @@ export const ShareRequestSchema = { title: 'ShareRequest' } as const; +export const StoredKeySchema = { + properties: { + key: { + type: 'string', + title: 'Key' + }, + size: { + type: 'integer', + title: 'Size' + }, + updated_at: { + type: 'string', + format: 'date-time', + title: 'Updated At' + } + }, + type: 'object', + required: ['key', 'size', 'updated_at'], + title: 'StoredKey' +} as const; + export const SubscriptionSchema = { properties: { endpoint: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 9428e45..1231dc1 100644 --- a/frontend/src/client/sdk.gen.ts +++ b/frontend/src/client/sdk.gen.ts @@ -3,7 +3,7 @@ import type { CancelablePromise } from './core/CancelablePromise'; import { OpenAPI } from './core/OpenAPI'; import { request as __request } from './core/request'; -import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; +import type { AiReadAiConfigResponse, AiUpdateAiConfigData, AiUpdateAiConfigResponse, AiDeleteAiConfigResponse, AiListAiModelsData, AiListAiModelsResponse, AiReadAiStatusResponse, AiReadControlNodesResponse, AiReadChatResponse, AiChatData, AiChatResponse, AiClearChatResponse, AlertsReadAlertsConfigResponse, AlertsSaveAlertsConfigData, AlertsSaveAlertsConfigResponse, AlertsReadWebpushKeyResponse, AlertsAddWebpushSubscriptionData, AlertsAddWebpushSubscriptionResponse, AlertsRemoveWebpushSubscriptionData, AlertsRemoveWebpushSubscriptionResponse, AlertsTestChannelData, AlertsTestChannelResponse, ArtifactsPutArtifactData, ArtifactsPutArtifactResponse, ArtifactsGetArtifactData, ArtifactsGetArtifactResponse, CloudReadStatusResponse, CloudEnrollData, CloudEnrollResponse, CloudAddRemoteUserData, CloudAddRemoteUserResponse, CloudDisconnectResponse, ConnectorsReadConnectorStateData, ConnectorsReadConnectorStateResponse, ConnectorsClearConnectorStateData, ConnectorsClearConnectorStateResponse, ConnectorsDeleteConnectorStateData, ConnectorsDeleteConnectorStateResponse, DashboardsReadDashboardsResponse, DashboardsReadDashboardData, DashboardsReadDashboardResponse, DashboardsCreateDashboardData, DashboardsCreateDashboardResponse, DashboardsSaveDashboardData, DashboardsSaveDashboardResponse, DashboardsDeleteDashboardData, DashboardsDeleteDashboardResponse, DashboardsGenerateResultsDashboardData, DashboardsGenerateResultsDashboardResponse, DashboardsPublishDashboardData, DashboardsPublishDashboardResponse, DashboardsDiscardDashboardDraftData, DashboardsDiscardDashboardDraftResponse, DashboardsRenameDashboardData, DashboardsRenameDashboardResponse, FlavorsReadFlavorsResponse, FlavorsCreateFlavorData, FlavorsCreateFlavorResponse, FlavorsUpdateFlavorData, FlavorsUpdateFlavorResponse, FlavorsDeleteFlavorData, FlavorsDeleteFlavorResponse, FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadGraphResponse, FlowsReadLibraryResponse, FlowsDeleteSharedNodeData, FlowsDeleteSharedNodeResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsPublishFlowData, FlowsPublishFlowResponse, FlowsDiscardDraftData, FlowsDiscardDraftResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsShareNodeData, FlowsShareNodeResponse, FlowsUnshareNodeData, FlowsUnshareNodeResponse, FlowsStartFlowData, FlowsStartFlowResponse, FlowsStopFlowData, FlowsStopFlowResponse, FlowsPauseFlowData, FlowsPauseFlowResponse, FlowsResumeFlowData, FlowsResumeFlowResponse, FlowsStepFlowData, FlowsStepFlowResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsCancelNodeData, FlowsCancelNodeResponse, FlowsAcknowledgeNodeErrorData, FlowsAcknowledgeNodeErrorResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, FlowsReadMessageHistoryData, FlowsReadMessageHistoryResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, MessagesReadMessagesResponse, MessagesPublishMessageData, MessagesPublishMessageResponse, MessagesReadMessageHistoryData, MessagesReadMessageHistoryResponse, ModulesReadModulesResponse, ModulesApplyModulesData, ModulesApplyModulesResponse, ModulesRefreshModulesResponse, OauthRegisterClientData, OauthRegisterClientResponse, OauthAuthorizeValidateData, OauthAuthorizeValidateResponse, OauthAuthorizeData, OauthAuthorizeResponse, OauthTokenData, OauthTokenResponse, OauthReadClientsResponse, OauthRevokeClientData, OauthRevokeClientResponse, ObservabilityReadSummaryResponse, ObservabilityReadTimeseriesData, ObservabilityReadTimeseriesResponse, ObservabilityReadFlowRollupsData, ObservabilityReadFlowRollupsResponse, ObservabilityReadRunsData, ObservabilityReadRunsResponse, ObservabilityReadEventsData, ObservabilityReadEventsResponse, ObservabilityReadDeadLettersData, ObservabilityReadDeadLettersResponse, PanelsReadPanelsResponse, PanelsSavePanelsData, PanelsSavePanelsResponse, PanelsStartPairingResponse, PanelsPollPairingData, PanelsPollPairingResponse, PanelsPendingDeviceData, PanelsPendingDeviceResponse, PanelsApprovePairingData, PanelsApprovePairingResponse, PanelsUnpairPanelData, PanelsUnpairPanelResponse, PanelsReadPanelData, PanelsReadPanelResponse, PrivateCreateUserData, PrivateCreateUserResponse, RunsCreateRunData, RunsCreateRunResponse, RunsCreateSweepData, RunsCreateSweepResponse, RunsImportRunsData, RunsImportRunsResponse, RunsReadRunsData, RunsReadRunsResponse, RunsDeleteRunsData, RunsDeleteRunsResponse, RunsReadOverviewResponse, RunsExportMetricsData, RunsExportMetricsResponse, RunsExportRunsData, RunsExportRunsResponse, RunsReadMetricNamesData, RunsReadMetricNamesResponse, RunsReadRunData, RunsReadRunResponse, RunsDeleteRunData, RunsDeleteRunResponse, RunsCancelRunData, RunsCancelRunResponse, RunsRetryRunData, RunsRetryRunResponse, RunsReadMetricsData, RunsReadMetricsResponse, RunsCompareMetricData, RunsCompareMetricResponse, SearchReadSearchIndexResponse, SecretsReadSecretsResponse, SecretsSaveSecretData, SecretsSaveSecretResponse, SecretsDeleteSecretData, SecretsDeleteSecretResponse, UsersReadUsersData, UsersReadUsersResponse, UsersCreateUserData, UsersCreateUserResponse, UsersReadUserMeResponse, UsersDeleteUserMeResponse, UsersUpdateUserMeData, UsersUpdateUserMeResponse, UsersUpdatePasswordMeData, UsersUpdatePasswordMeResponse, UsersRegisterUserData, UsersRegisterUserResponse, UsersReadUserByIdData, UsersReadUserByIdResponse, UsersUpdateUserData, UsersUpdateUserResponse, UsersDeleteUserData, UsersDeleteUserResponse, UtilsTestEmailData, UtilsTestEmailResponse, UtilsReadConfigResponse, UtilsHealthCheckResponse, UtilsHealthResponse, WorkersReadWorkersResponse, WorkersReadResourcesResponse, WorkersIssueTokenData, WorkersIssueTokenResponse, WorkersReadRuntimeResponse } from './types.gen'; export class AiService { /** @@ -411,6 +411,73 @@ export class CloudService { } } +export class ConnectorsService { + /** + * Read Connector State + * The keys this connector keeps, without their values. + * @param data The data for the request. + * @param data.nodeType + * @returns ConnectorStatePublic Successful Response + * @throws ApiError + */ + public static readConnectorState(data: ConnectorsReadConnectorStateData): CancelablePromise { + return __request(OpenAPI, { + method: 'GET', + url: '/api/v1/connectors/{node_type}/state', + path: { + node_type: data.nodeType + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Clear Connector State + * Forget everything this connector keeps. It starts over on next use. + * @param data The data for the request. + * @param data.nodeType + * @returns Message Successful Response + * @throws ApiError + */ + public static clearConnectorState(data: ConnectorsClearConnectorStateData): CancelablePromise { + return __request(OpenAPI, { + method: 'DELETE', + url: '/api/v1/connectors/{node_type}/state', + path: { + node_type: data.nodeType + }, + errors: { + 422: 'Validation Error' + } + }); + } + + /** + * Delete Connector State + * Forget one key, such as the certificate pinned for one bridge. + * @param data The data for the request. + * @param data.nodeType + * @param data.key + * @returns Message Successful Response + * @throws ApiError + */ + public static deleteConnectorState(data: ConnectorsDeleteConnectorStateData): CancelablePromise { + return __request(OpenAPI, { + method: 'DELETE', + url: '/api/v1/connectors/{node_type}/state/{key}', + path: { + node_type: data.nodeType, + key: data.key + }, + errors: { + 422: 'Validation Error' + } + }); + } +} + export class DashboardsService { /** * Read Dashboards diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index f302818..09e4c93 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -142,6 +142,12 @@ export type ChatItem = { error?: (string | null); }; +export type ConnectorStatePublic = { + connector: string; + data: Array; + count: number; +}; + /** * A wren node the chat can talk to: where it sends and where it reads. */ @@ -1235,6 +1241,12 @@ export type ShareRequest = { lib_name: string; }; +export type StoredKey = { + key: string; + size: number; + updated_at: string; +}; + /** * What a browser handed us. Opaque apart from the endpoint's host. */ @@ -1553,6 +1565,25 @@ export type CloudAddRemoteUserResponse = (UserPublic); export type CloudDisconnectResponse = (Message); +export type ConnectorsReadConnectorStateData = { + nodeType: string; +}; + +export type ConnectorsReadConnectorStateResponse = (ConnectorStatePublic); + +export type ConnectorsClearConnectorStateData = { + nodeType: string; +}; + +export type ConnectorsClearConnectorStateResponse = (Message); + +export type ConnectorsDeleteConnectorStateData = { + key: string; + nodeType: string; +}; + +export type ConnectorsDeleteConnectorStateResponse = (Message); + export type DashboardsReadDashboardsResponse = (DashboardsPublic); export type DashboardsReadDashboardData = { diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index a49e6f6..237cf30 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -16,6 +16,7 @@ import { } from "react" import { + ConnectorsService, type DType, type FlowOrigin, FlowsService, @@ -1233,6 +1234,84 @@ const SCAFFOLD_SHAPE = (() => { ) })() +/** + * What a connector keeps across restarts, such as a certificate it pinned on + * first contact. Clearing it is how a replaced device gets trusted again. + */ +function StoredStateSection({ nodeType }: { nodeType: string }) { + const queryClient = useQueryClient() + const { showSuccessToast, showErrorToast } = useCustomToast() + const [confirming, setConfirming] = useState(false) + const queryKey = ["connector-state", nodeType] + const { data } = useQuery({ + queryKey, + queryFn: () => ConnectorsService.readConnectorState({ nodeType }), + }) + const clear = useMutation({ + mutationFn: () => ConnectorsService.clearConnectorState({ nodeType }), + onSuccess: () => { + showSuccessToast("Stored state cleared") + setConfirming(false) + queryClient.invalidateQueries({ queryKey }) + }, + onError: () => showErrorToast("The stored state could not be cleared."), + }) + + if (!data) return null + return ( +
+
+ Stored state + + What {data.connector} keeps across restarts, such as a certificate + pinned on first contact. Every node of this connector shares it, and + clearing it makes them start over. + +
+
+ + {data.count === 0 + ? "Nothing stored" + : `${data.count} key${data.count === 1 ? "" : "s"}`} + + {confirming ? ( +
+ + +
+ ) : ( + + )} +
+
+ ) +} + /** The name the node function sees; the engine derives it the same way. */ const portName = (spec: MessageSpec) => spec.port || (spec.name ?? "").split(".").pop() || "" @@ -1487,6 +1566,9 @@ function PanelBody({ onChange={(params) => editNode({ ...node, params })} /> ) : null} + {nodeType?.plugin && user?.is_superuser ? ( + + ) : null} {hasSource ? (