diff --git a/backend/app/api/routes/flows.py b/backend/app/api/routes/flows.py index b991c0f..f3d3ad1 100644 --- a/backend/app/api/routes/flows.py +++ b/backend/app/api/routes/flows.py @@ -21,6 +21,7 @@ from app.flow.events import event_bus from app.flow.messages import qualify from app.flow.pipeline import ValidationIssue from app.flow.schemas import ( + NAME_PATTERN, FlowDef, FlowsPublic, FlowStatePublic, @@ -30,7 +31,7 @@ from app.flow.schemas import ( NodeStatusPublic, NodeTypeInfo, ) -from app.flow.store import FlowNotFound +from app.flow.store import FlowExists, FlowNotFound from app.models import Message router = APIRouter( @@ -53,6 +54,10 @@ class ValidationResult(BaseModel): issues: list[ValidationIssue] = [] +class RenameRequest(BaseModel): + new_name: str + + class RunRequest(BaseModel): inputs: dict[str, Any] = {} @@ -154,6 +159,33 @@ async def delete_flow(name: str, controller: FlowControllerDep) -> Any: 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)) + + await controller.reload() + return _detail(controller, renamed) + + # ----------------------------------------------------------------------------- # Node source # ----------------------------------------------------------------------------- diff --git a/backend/app/flow/pipeline.py b/backend/app/flow/pipeline.py index efd8433..0cdfcc0 100644 --- a/backend/app/flow/pipeline.py +++ b/backend/app/flow/pipeline.py @@ -338,6 +338,9 @@ class Pipeline: "type": "node_executed", "flow": node.flow, "node": node.id, + # A node that returns nothing ran but published nothing, + # which is a different thing to show than one that emitted. + "outputs": len(result or {}), "duration_ms": round((time.perf_counter() - started) * 1000, 2), "ts": time.time(), } @@ -443,6 +446,18 @@ class Pipeline: "ts": ts, } ) + # An injecting node — an MQTT subscriber, a webhook — publishes + # without going through the executor, but it did emit. + self._publish( + { + "type": "node_executed", + "flow": node.flow, + "node": node.id, + "outputs": len(outputs), + "duration_ms": 0, + "ts": ts, + } + ) downstream = set(self._get_downstream(node)) if not downstream: diff --git a/backend/app/flow/store.py b/backend/app/flow/store.py index 486c9d7..775d139 100644 --- a/backend/app/flow/store.py +++ b/backend/app/flow/store.py @@ -34,6 +34,15 @@ class FlowNotFound(KeyError): return f"No flow named '{self.name}'" +class FlowExists(ValueError): + def __init__(self, name: str) -> None: + super().__init__(name) + self.name = name + + def __str__(self) -> str: + return f"There is already a flow named '{self.name}'" + + class FlowStore: """Reads and writes flows, committing every change.""" @@ -129,6 +138,50 @@ class FlowStore: shutil.rmtree(directory) self._commit(f"Delete flow '{name}'") + def rename_flow(self, name: str, new_name: str) -> FlowDef: + """Rename a flow, carrying its nodes and any references to it. + + A flow's name is the namespace of its messages, so other flows reading + ``old.temperature`` are rewritten to read ``new.temperature`` — leaving + them pointing at a flow that no longer exists would break them silently. + """ + if not self.exists(name): + raise FlowNotFound(name) + if self.exists(new_name): + raise FlowExists(new_name) + + flow = self.read_flow(name) + self._flow_dir(name).rename(self._flow_dir(new_name)) + + renamed = flow.model_copy(update={"name": new_name}) + self._flow_file(new_name).write_text(renamed.model_dump_json(indent=2) + "\n") + + for other in self.read_all(): + if other.name == new_name: + continue + if self._retarget(other, f"{name}.", f"{new_name}."): + self._flow_file(other.name).write_text( + other.model_dump_json(indent=2) + "\n" + ) + + self._commit(f"Rename flow '{name}' to '{new_name}'") + return renamed + + @staticmethod + def _retarget(flow: FlowDef, old_prefix: str, new_prefix: str) -> bool: + """Point this flow's cross-flow message names at a renamed flow.""" + changed = False + for node in flow.nodes: + for specs in (node.requires, node.provides): + for position, spec in enumerate(specs): + if spec.name.startswith(old_prefix): + tail = spec.name[len(old_prefix) :] + specs[position] = spec.model_copy( + update={"name": new_prefix + tail} + ) + changed = True + return changed + # ------------------------------------------------------------------------- # Node source # ------------------------------------------------------------------------- diff --git a/backend/tests/api/routes/test_flows.py b/backend/tests/api/routes/test_flows.py index b5abb0e..7a40922 100644 --- a/backend/tests/api/routes/test_flows.py +++ b/backend/tests/api/routes/test_flows.py @@ -144,3 +144,65 @@ def test_node_types_are_listed( by_type = {entry["type"]: entry for entry in types} assert by_type["python"]["has_source"] is True assert "properties" in by_type["mqtt"]["params_schema"] + + +def test_rename_flow( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow()) + + response = client.post( + f"{PREFIX}/demo/rename", + headers=superuser_token_headers, + json={"new_name": "demo_renamed"}, + ) + assert response.status_code == 200 + assert response.json()["definition"]["name"] == "demo_renamed" + + assert ( + client.get(f"{PREFIX}/demo", headers=superuser_token_headers).status_code == 404 + ) + assert ( + client.get( + f"{PREFIX}/demo_renamed", headers=superuser_token_headers + ).status_code + == 200 + ) + + # Put it back so the tests that follow find the flow they expect. + client.post( + f"{PREFIX}/demo_renamed/rename", + headers=superuser_token_headers, + json={"new_name": "demo"}, + ) + + +def test_rename_onto_a_taken_name_is_refused( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow()) + client.put( + f"{PREFIX}/occupied", headers=superuser_token_headers, json=a_flow("occupied") + ) + + response = client.post( + f"{PREFIX}/demo/rename", + headers=superuser_token_headers, + json={"new_name": "occupied"}, + ) + assert response.status_code == 409 + + client.delete(f"{PREFIX}/occupied", headers=superuser_token_headers) + + +def test_rename_rejects_an_invalid_name( + client: TestClient, superuser_token_headers: dict[str, str] +) -> None: + client.put(f"{PREFIX}/demo", headers=superuser_token_headers, json=a_flow()) + + response = client.post( + f"{PREFIX}/demo/rename", + headers=superuser_token_headers, + json={"new_name": "Not A Flow Name"}, + ) + assert response.status_code == 400 diff --git a/backend/tests/flow/test_store.py b/backend/tests/flow/test_store.py index 0beee3f..b92541c 100644 --- a/backend/tests/flow/test_store.py +++ b/backend/tests/flow/test_store.py @@ -5,7 +5,7 @@ import pytest from app.flow.messages import MessageSpec from app.flow.schemas import FlowDef, NodeDef -from app.flow.store import FlowNotFound, FlowStore +from app.flow.store import FlowExists, FlowNotFound, FlowStore @pytest.fixture @@ -73,3 +73,47 @@ def test_deleting_removes_flow_and_its_nodes(store: FlowStore): assert store.list_flows() == [] assert not (store.root / "heating").exists() + + +def test_renaming_a_flow_carries_its_nodes(store: FlowStore): + store.write_flow(a_flow()) + store.write_node_source( + "heating", "sensor", "def process(params):\n return {}\n" + ) + + renamed = store.rename_flow("heating", "warmth") + + assert renamed.name == "warmth" + assert store.list_flows() == ["warmth"] + assert "def process" in store.read_node_source("warmth", "sensor") + + +def test_renaming_a_flow_repoints_the_flows_reading_it(store: FlowStore): + store.write_flow(a_flow()) + store.write_flow( + FlowDef( + name="display", + nodes=[ + NodeDef( + id="gauge", + # Reads across the flow boundary, so the name must follow. + requires=[MessageSpec(name="heating.temp")], + ) + ], + ) + ) + + store.rename_flow("heating", "warmth") + + display = store.read_flow("display") + assert display.nodes[0].requires[0].name == "warmth.temp" + + +def test_renaming_onto_an_existing_name_is_refused(store: FlowStore): + store.write_flow(a_flow()) + store.write_flow(FlowDef(name="warmth")) + + with pytest.raises(FlowExists): + store.rename_flow("heating", "warmth") + + assert store.list_flows() == ["heating", "warmth"] diff --git a/frontend/src/client/schemas.gen.ts b/frontend/src/client/schemas.gen.ts index c8fc769..39089d7 100644 --- a/frontend/src/client/schemas.gen.ts +++ b/frontend/src/client/schemas.gen.ts @@ -586,6 +586,18 @@ export const PrivateUserCreateSchema = { title: 'PrivateUserCreate' } as const; +export const RenameRequestSchema = { + properties: { + new_name: { + type: 'string', + title: 'New Name' + } + }, + type: 'object', + required: ['new_name'], + title: 'RenameRequest' +} as const; + export const RunRequestSchema = { properties: { inputs: { diff --git a/frontend/src/client/sdk.gen.ts b/frontend/src/client/sdk.gen.ts index 7a171c6..3333676 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 { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, 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, UtilsHealthCheckResponse } from './types.gen'; +import type { FlowsReadFlowsResponse, FlowsReadNodeTypesResponse, FlowsReadFlowData, FlowsReadFlowResponse, FlowsSaveFlowData, FlowsSaveFlowResponse, FlowsDeleteFlowData, FlowsDeleteFlowResponse, FlowsRenameFlowData, FlowsRenameFlowResponse, FlowsReadNodeSourceData, FlowsReadNodeSourceResponse, FlowsSaveNodeSourceData, FlowsSaveNodeSourceResponse, FlowsValidateFlowData, FlowsValidateFlowResponse, FlowsRunFlowData, FlowsRunFlowResponse, FlowsTriggerNodeData, FlowsTriggerNodeResponse, FlowsReadFlowStateData, FlowsReadFlowStateResponse, LoginLoginAccessTokenData, LoginLoginAccessTokenResponse, LoginTestTokenResponse, LoginRecoverPasswordData, LoginRecoverPasswordResponse, LoginResetPasswordData, LoginResetPasswordResponse, LoginRecoverPasswordHtmlContentData, LoginRecoverPasswordHtmlContentResponse, PrivateCreateUserData, PrivateCreateUserResponse, 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, UtilsHealthCheckResponse } from './types.gen'; export class FlowsService { /** @@ -98,6 +98,30 @@ export class FlowsService { }); } + /** + * Rename Flow + * Rename a flow, along with every reference to its messages. + * @param data The data for the request. + * @param data.name + * @param data.requestBody + * @returns FlowDetail Successful Response + * @throws ApiError + */ + public static renameFlow(data: FlowsRenameFlowData): CancelablePromise { + return __request(OpenAPI, { + method: 'POST', + url: '/api/v1/flows/{name}/rename', + path: { + name: data.name + }, + body: data.requestBody, + mediaType: 'application/json', + errors: { + 422: 'Validation Error' + } + }); + } + /** * Read Node Source * Read a node's Python source. diff --git a/frontend/src/client/types.gen.ts b/frontend/src/client/types.gen.ts index bf67110..e8269b2 100644 --- a/frontend/src/client/types.gen.ts +++ b/frontend/src/client/types.gen.ts @@ -193,6 +193,10 @@ export type PrivateUserCreate = { is_verified?: boolean; }; +export type RenameRequest = { + new_name: string; +}; + export type RunRequest = { inputs?: { [key: string]: unknown; @@ -313,6 +317,13 @@ export type FlowsDeleteFlowData = { export type FlowsDeleteFlowResponse = (Message); +export type FlowsRenameFlowData = { + name: string; + requestBody: RenameRequest; +}; + +export type FlowsRenameFlowResponse = (FlowDetail); + export type FlowsReadNodeSourceData = { name: string; nodeId: string; diff --git a/frontend/src/components/Flow/FlowDock.tsx b/frontend/src/components/Flow/FlowDock.tsx index acbe2af..2d0bd6c 100644 --- a/frontend/src/components/Flow/FlowDock.tsx +++ b/frontend/src/components/Flow/FlowDock.tsx @@ -3,6 +3,7 @@ import { AlertCircle, Loader2, Maximize2, + Pencil, Play, Plus, ZoomIn, @@ -33,12 +34,14 @@ export function FlowDock({ issues, running, onAddNode, + onEditFlow, onRun, onFocusNode, }: { issues: ValidationIssue[] running: boolean onAddNode: () => void + onEditFlow: () => void onRun: () => void onFocusNode: (nodeId: string) => void }) { @@ -68,6 +71,22 @@ export function FlowDock({ Add a node (⌘K) + + + + + Flow settings + + + } + > +
+
+ Name +
+ setName(event.target.value)} + onKeyDown={(event) => { + if (event.key === "Enter" && valid && changed) { + onRename(name) + } + }} + /> + +
+

+ {valid || !name + ? "Messages in this flow are named after it, so other flows reading them follow the rename." + : "Lowercase letters, digits and underscores, starting with a letter."} +

+
+ +
+ Contents +

+ {nodeCount === 0 + ? "No nodes yet." + : `${nodeCount} node${nodeCount === 1 ? "" : "s"}.`} +

+
+
+ + + + + + + Delete {definition.title || definition.name}? + + + This removes the flow and the code of its{" "} + {nodeCount === 1 ? "node" : `${nodeCount} nodes`}. Its history + stays in the flow store's git repository. + + + + + + + + + + ) +} diff --git a/frontend/src/components/Flow/NodePanel.tsx b/frontend/src/components/Flow/NodePanel.tsx index b4364b3..881aa10 100644 --- a/frontend/src/components/Flow/NodePanel.tsx +++ b/frontend/src/components/Flow/NodePanel.tsx @@ -1,6 +1,5 @@ import { useQuery } from "@tanstack/react-query" import { X } from "lucide-react" -import { AnimatePresence, motion } from "motion/react" import { lazy, Suspense, useEffect, useRef, useState } from "react" import type { DType, MessageSpec, NodeDef_Input, NodeTypeInfo } from "@/client" @@ -22,33 +21,15 @@ import { SelectTrigger, SelectValue, } from "@/components/ui/select" -import { Sheet, SheetContent, SheetTitle } from "@/components/ui/sheet" import { Switch } from "@/components/ui/switch" -import { useIsMobile } from "@/hooks/useMobile" -import { duration, easeEmphasized, easeStandard } from "@/lib/motion" import { nodeSourceQueryOptions } from "./queries" +import { PANEL_SECTION, SidePanel } from "./SidePanel" const NodeEditor = lazy(() => import("./NodeEditor")) const DTYPES: DType[] = ["float", "int", "str", "bool", "json"] -const SECTION = - "text-xs font-medium uppercase tracking-[0.5px] text-muted-foreground" - -/** Same grammar as the shared `slideUp`, on the axis this panel travels. */ -const panelSlide = { - hidden: { opacity: 0, x: 16 }, - visible: { - opacity: 1, - x: 0, - transition: { duration: duration.base, ease: easeEmphasized }, - }, - exit: { - opacity: 0, - x: 16, - transition: { duration: duration.fast, ease: easeStandard }, - }, -} +const SECTION = PANEL_SECTION /** * A message name, typed freely or picked from the names already in play. @@ -297,8 +278,6 @@ function PanelBody({ suggestions, onChange, onSaveSource, - onClose, - onDelete, }: { node: NodeDef_Input flow: string @@ -306,8 +285,6 @@ function PanelBody({ suggestions: PortSuggestions onChange: (next: NodeDef_Input) => void onSaveSource: (code: string) => void - onClose: () => void - onDelete: () => void }) { const hasSource = nodeType?.has_source ?? node.type === "python" const { data: source } = useQuery({ @@ -343,78 +320,47 @@ function PanelBody({ return ( <> -
- onChange({ ...node, title: event.target.value })} +
+ onChange({ ...node, requires })} + /> + onChange({ ...node, provides })} + /> + onChange({ ...node, params })} /> -
-
-
- onChange({ ...node, requires })} - /> - onChange({ ...node, provides })} - /> - onChange({ ...node, params })} - /> -
- - {hasSource ? ( -
- Code -
- - } - > - - -
+ {hasSource ? ( +
+ Code +
+ + } + > + +
- ) : null} -
- -
- -
+
+ ) : null} ) } @@ -445,73 +391,48 @@ export function NodePanel({ onClose: () => void onDelete: () => void }) { - const isMobile = useIsMobile() const nodeType = nodeTypes.find((entry) => entry.type === node?.type) - useEffect(() => { - if (!node) return - const onKey = (event: KeyboardEvent) => { - if (event.key === "Escape") onClose() - } - window.addEventListener("keydown", onKey) - return () => window.removeEventListener("keydown", onKey) - }, [node, onClose]) - - if (isMobile) { - return ( - !open && onClose()}> - event.preventDefault()} - > - Node settings - {node ? ( - - ) : null} - - - ) - } - return ( - - {node ? ( - - + onChange({ ...node, title: event.target.value }) + } /> - + ) : null + } + footer={ + + } + > + {node ? ( + ) : null} - + ) } diff --git a/frontend/src/components/Flow/SidePanel.tsx b/frontend/src/components/Flow/SidePanel.tsx new file mode 100644 index 0000000..f0a1e25 --- /dev/null +++ b/frontend/src/components/Flow/SidePanel.tsx @@ -0,0 +1,135 @@ +import { X } from "lucide-react" +import { AnimatePresence, motion } from "motion/react" +import type { ReactNode } from "react" +import { useEffect } from "react" + +import { Button } from "@/components/ui/button" +import { Sheet, SheetContent, SheetTitle } from "@/components/ui/sheet" +import { useIsMobile } from "@/hooks/useMobile" +import { duration, easeEmphasized, easeStandard } from "@/lib/motion" + +/** Same grammar as the shared `slideUp`, on the axis this panel travels. */ +const panelSlide = { + hidden: { opacity: 0, x: 16 }, + visible: { + opacity: 1, + x: 0, + transition: { duration: duration.base, ease: easeEmphasized }, + }, + exit: { + opacity: 0, + x: 16, + transition: { duration: duration.fast, ease: easeStandard }, + }, +} + +export const PANEL_SECTION = + "text-xs font-medium uppercase tracking-[0.5px] text-muted-foreground" + +/** + * The editor's settings panel: floating over the canvas so the graph stays + * visible and running behind it, a full-screen sheet where there is no room + * for that. + * + * Node settings and flow settings share it, so the two read as one surface. + */ +export function SidePanel({ + open, + label, + testId, + bodyKey, + header, + footer, + children, + onClose, +}: { + open: boolean + /** Names the panel for screen readers. */ + label: string + testId: string + /** Remounts the contents when the thing being edited changes. */ + bodyKey: string + header: ReactNode + footer?: ReactNode + children: ReactNode + onClose: () => void +}) { + const isMobile = useIsMobile() + + useEffect(() => { + if (!open) return + const onKey = (event: KeyboardEvent) => { + if (event.key === "Escape") onClose() + } + window.addEventListener("keydown", onKey) + return () => window.removeEventListener("keydown", onKey) + }, [open, onClose]) + + const contents = ( + <> +
+ {header} + +
+ +
+ {children} +
+ + {footer ? ( +
+ {footer} +
+ ) : null} + + ) + + if (isMobile) { + return ( + !next && onClose()}> + event.preventDefault()} + > + {label} + {open ? ( +
+ {contents} +
+ ) : null} + + + ) + } + + return ( + + {open ? ( + + {contents} + + ) : null} + + ) +} diff --git a/frontend/src/components/Flow/flow.css b/frontend/src/components/Flow/flow.css index edec065..ba8220c 100644 --- a/frontend/src/components/Flow/flow.css +++ b/frontend/src/components/Flow/flow.css @@ -49,6 +49,30 @@ } } +/* A node that just published something says so, once, and settles. */ +@media (prefers-reduced-motion: no-preference) { + .node-pulse { + position: absolute; + inset: -3px; + border-radius: inherit; + border: 2px solid var(--primary); + box-shadow: 0 0 0 3px color-mix(in srgb, var(--primary) 18%, transparent); + pointer-events: none; + animation: node-pulse var(--duration-slow) var(--ease-emphasized) forwards; + } + + @keyframes node-pulse { + from { + opacity: 1; + transform: scale(1); + } + to { + opacity: 0; + transform: scale(1.09); + } + } +} + /* Handles are neutral: the one brand-secondary affordance here is Run. */ .react-flow__handle { width: 12px; diff --git a/frontend/src/components/Flow/liveStore.ts b/frontend/src/components/Flow/liveStore.ts index a0294e5..03da5a7 100644 --- a/frontend/src/components/Flow/liveStore.ts +++ b/frontend/src/components/Flow/liveStore.ts @@ -18,6 +18,9 @@ type Listener = () => void const values = new Map() const statuses = new Map() +// How many times a node has emitted. The number itself means nothing; a change +// is what restarts the pulse. +const emits = new Map() const listeners = new Map>() let connected = false @@ -72,6 +75,10 @@ export const liveStore = { getStatus(nodeId: string) { return statuses.get(nodeId) }, + recordEmit(nodeId: string) { + emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1) + notify(`emit:${nodeId}`) + }, setConnected(next: boolean) { if (connected === next) return connected = next @@ -85,6 +92,8 @@ export const liveStore = { values.clear() for (const key of statuses.keys()) notify(`status:${key}`) statuses.clear() + for (const key of emits.keys()) notify(`emit:${key}`) + emits.clear() }, } @@ -102,6 +111,14 @@ export function useNodeStatus(nodeId: string): LiveStatus | undefined { ) } +/** Increments each time the node publishes something. */ +export function useNodeEmits(nodeId: string): number { + return useSyncExternalStore( + (listener) => subscribeKey(`emit:${nodeId}`, listener), + () => emits.get(nodeId) ?? 0, + ) +} + export function useLiveConnection(): boolean { return useSyncExternalStore( (listener) => { diff --git a/frontend/src/components/Flow/useFlowSocket.ts b/frontend/src/components/Flow/useFlowSocket.ts index fe2375f..ac37171 100644 --- a/frontend/src/components/Flow/useFlowSocket.ts +++ b/frontend/src/components/Flow/useFlowSocket.ts @@ -13,7 +13,7 @@ type FlowEvent = nodes: { id: string; status: string; error?: string | null }[] } | { type: "message_value"; name: string; value: unknown; ts: number } - | { type: "node_executed"; node: string } + | { type: "node_executed"; node: string; outputs: number } | { type: "node_error"; node: string; error: string } | { type: "node_status"; node: string; status: string; error?: string | null } | { @@ -71,6 +71,7 @@ export function useFlowSocket(onAuthFailure?: () => void): void { break case "node_executed": liveStore.setStatus(message.node, { status: "success" }) + if (message.outputs > 0) liveStore.recordEmit(message.node) break case "node_error": liveStore.setStatus(message.node, { diff --git a/frontend/src/index.css b/frontend/src/index.css index 2bdbe28..b8307e7 100644 --- a/frontend/src/index.css +++ b/frontend/src/index.css @@ -35,8 +35,8 @@ --ease-emphasized: cubic-bezier(0.2, 0, 0, 1); --ease-standard: cubic-bezier(0.4, 0, 0.2, 1); --duration-fast: 150ms; - --duration-base: 200ms; - --duration-slow: 300ms; + --duration-base: 250ms; + --duration-slow: 500ms; --color-background: var(--background); --color-foreground: var(--foreground); --color-card: var(--card);