diff --git a/packages/client/src/pages/flow/edit.tsx b/packages/client/src/pages/flow/edit.tsx
index fc58004c7..d0bccf96c 100644
--- a/packages/client/src/pages/flow/edit.tsx
+++ b/packages/client/src/pages/flow/edit.tsx
@@ -35,6 +35,7 @@ import {
NodeCollectionSchema,
NodeGraphQLCollectionSchema,
NodeHttpCollectionSchema,
+ NodeWsConnectionCollectionSchema,
} from '@the-dev-tools/spec/tanstack-db/v1/api/flow';
import { Button, ButtonAsRouteLink } from '@the-dev-tools/ui/button';
import { Checkbox } from '@the-dev-tools/ui/checkbox';
@@ -83,6 +84,9 @@ import { GraphQLNode, GraphQLSettings } from './nodes/graphql';
import { HttpNode, HttpSettings } from './nodes/http';
import { JavaScriptNode, JavaScriptSettings } from './nodes/javascript';
import { ManualStartNode } from './nodes/manual-start';
+import { WaitNode, WaitSettings } from './nodes/wait';
+import { WsConnectionNode, WsConnectionSettings } from './nodes/ws-connection';
+import { WsSendNode, WsSendSettings } from './nodes/ws-send';
import { useFlowSelection } from './selection';
import { useUndoStack } from './undo';
import { useViewport, VIEWPORT_MAX_ZOOM, VIEWPORT_MIN_ZOOM } from './viewport';
@@ -99,6 +103,9 @@ export const nodeTypes: XF.NodeTypes = {
[NodeKind.JS]: JavaScriptNode,
[NodeKind.MANUAL_START]: ManualStartNode,
[NodeKind.UNSPECIFIED]: () => null,
+ [NodeKind.WAIT]: WaitNode,
+ [NodeKind.WS_CONNECTION]: WsConnectionNode,
+ [NodeKind.WS_SEND]: WsSendNode,
};
export const FlowEditPage = () => {
@@ -153,6 +160,7 @@ export const Flow = ({ children }: PropsWithChildren) => {
const nodeCollection = useApiCollection(NodeCollectionSchema);
const nodeGraphQLCollection = useApiCollection(NodeGraphQLCollectionSchema);
const nodeHttpCollection = useApiCollection(NodeHttpCollectionSchema);
+ const nodeWsConnectionCollection = useApiCollection(NodeWsConnectionCollectionSchema);
const nodeEditDialog = useNodeEditDialog();
@@ -461,6 +469,23 @@ export const Flow = ({ children }: PropsWithChildren) => {
position,
});
}
+
+ if (file?.kind === FileKind.WEB_SOCKET) {
+ const nodeId = Ulid.generate().bytes;
+
+ nodeWsConnectionCollection.utils.insert({
+ nodeId,
+ websocketId: file.fileId,
+ });
+
+ nodeCollection.utils.insert({
+ flowId,
+ kind: NodeKind.WS_CONNECTION,
+ name: `ws_connection_${getNodes().length}`,
+ nodeId,
+ position,
+ });
+ }
},
ref,
});
@@ -881,6 +906,9 @@ const useNodeEditDialog = () => {
Match.when({ kind: NodeKind.AI }, (_) =>
),
Match.when({ kind: NodeKind.AI_PROVIDER }, (_) =>
),
Match.when({ kind: NodeKind.AI_MEMORY }, (_) =>
),
+ Match.when({ kind: NodeKind.WAIT }, () =>
),
+ Match.when({ kind: NodeKind.WS_CONNECTION }, () =>
),
+ Match.when({ kind: NodeKind.WS_SEND }, () =>
),
Match.orElse(() => null),
);
diff --git a/packages/client/src/pages/flow/handle.tsx b/packages/client/src/pages/flow/handle.tsx
index e85a4e763..48e47bd36 100644
--- a/packages/client/src/pages/flow/handle.tsx
+++ b/packages/client/src/pages/flow/handle.tsx
@@ -16,6 +16,7 @@ import { FlowContext } from './context';
interface HandleProps extends Omit
{
alwaysVisible?: boolean;
kind?: HandleKind;
+ label?: string;
nodeId: Uint8Array;
nodeOffset?: { x?: number; y?: number };
Sidebar?: (props: AddNodeSidebarProps) => ReactNode;
@@ -25,6 +26,7 @@ export const Handle = ({
alwaysVisible,
className,
kind = HandleKind.UNSPECIFIED,
+ label: labelOverride,
nodeId,
nodeOffset,
Sidebar = AddNodeSidebar,
@@ -38,16 +40,19 @@ export const Handle = ({
const id = kind === HandleKind.UNSPECIFIED ? null : kind.toString();
- const label = pipe(
- Match.value(kind),
- Match.when(HandleKind.ELSE, () => 'Else'),
- Match.when(HandleKind.THEN, () => 'Then'),
- Match.when(HandleKind.LOOP, () => 'Loop'),
- Match.when(HandleKind.AI_PROVIDER, () => 'Provider'),
- Match.when(HandleKind.AI_MEMORY, () => 'Memory'),
- Match.when(HandleKind.AI_TOOLS, () => 'Tools'),
- Match.orElse(() => null),
- );
+ const label =
+ labelOverride ??
+ pipe(
+ Match.value(kind),
+ Match.when(HandleKind.ELSE, () => 'Else'),
+ Match.when(HandleKind.THEN, () => 'Then'),
+ Match.when(HandleKind.LOOP, () => 'Loop'),
+ Match.when(HandleKind.AI_PROVIDER, () => 'Provider'),
+ Match.when(HandleKind.AI_MEMORY, () => 'Memory'),
+ Match.when(HandleKind.AI_TOOLS, () => 'Tools'),
+ Match.when(HandleKind.WS_MESSAGE, () => 'Message'),
+ Match.orElse(() => null),
+ );
const edgeCollection = useApiCollection(EdgeCollectionSchema);
diff --git a/packages/client/src/pages/flow/node.tsx b/packages/client/src/pages/flow/node.tsx
index 4fb6b6ad8..f431f88ef 100644
--- a/packages/client/src/pages/flow/node.tsx
+++ b/packages/client/src/pages/flow/node.tsx
@@ -9,7 +9,7 @@ import {
usePacedMutations,
} from '@tanstack/react-db';
import * as XF from '@xyflow/react';
-import { Array, Match, Option, pipe, Schema } from 'effect';
+import { Array, Match, pipe, Schema } from 'effect';
import { Ulid } from 'id128';
import { ReactNode, useCallback, useContext, useRef, useState } from 'react';
import { Button as AriaButton, Key, Tooltip, TooltipTrigger, Tree } from 'react-aria-components';
@@ -454,46 +454,42 @@ interface NodeSettingsBodyProps {
export const NodeSettingsBody = ({ children, input, nodeId, output, settingsHeader, title }: NodeSettingsBodyProps) => {
const executionCollection = useApiCollection(NodeExecutionCollectionSchema);
- const { data: executions } = useLiveQuery(
+ const { data: rawExecutions } = useLiveQuery(
(_) => {
const item = _.from({ item: executionCollection })
.where((_) => eq(_.item.nodeId, nodeId))
.fn.select((_) => ({
...pick(_.item, 'nodeExecutionId', 'name'),
- order: Ulid.construct(_.item.nodeExecutionId).toCanonical(),
+ key: Ulid.construct(_.item.nodeExecutionId).toCanonical(),
}));
- return _.from({ item }).orderBy((_) => _.item.order, 'desc');
+ return _.from({ item }).orderBy((_) => _.item.key, 'desc');
},
[executionCollection, nodeId],
);
- const latestExecutionId = pipe(
- Array.head(executions),
- Option.map((_) => _.nodeExecutionId),
- Option.getOrNull,
- );
+ // Deduplicate by canonical ULID — reactive queries can transiently emit duplicates during sync
+ const executions = Array.dedupeWith(rawExecutions, (a, b) => a.key === b.key);
- const latestExecutionIdCan = latestExecutionId ? Ulid.construct(latestExecutionId).toCanonical() : null;
+ const latestKey = executions[0]?.key ?? null;
- const [selectedExecKey, setSelectedExecKey] = useState(latestExecutionId ? 'latest' : null);
+ // Track whether user is "following" the latest execution (auto-advance on new arrivals)
+ // vs pinned to a specific historical execution.
+ const [pinnedKey, setPinnedKey] = useState(null);
+ const selectedKey = pinnedKey ?? latestKey;
- const selectedExecutionId =
- selectedExecKey === 'latest'
- ? latestExecutionId
- : typeof selectedExecKey === 'string'
- ? Ulid.fromCanonical(selectedExecKey).bytes
- : null;
+ const handleSelectionChange = (key: Key | null) => {
+ // If user selects the latest execution, clear the pin so we follow new arrivals
+ setPinnedKey(key === latestKey ? null : typeof key === 'string' ? key : null);
+ };
- const selectedExecutionIdCan = selectedExecutionId ? Ulid.construct(selectedExecutionId).toCanonical() : null;
+ const selectedExecutionId =
+ typeof selectedKey === 'string' ? (executions.find((_) => _.key === selectedKey)?.nodeExecutionId ?? null) : null;
- // Fix React Aria over-rendering non-visible components
+ // React Aria workaround: only render full list when dropdown is open
// https://github.com/adobe/react-spectrum/issues/8783#issuecomment-3233350825
- // TODO: move the workaround to an improved select component
const [isExecListOpen, setIsExecListOpen] = useState(false);
- const execItems = isExecListOpen
- ? executions
- : executions.filter((_) => Ulid.construct(_.nodeExecutionId).toCanonical() === selectedExecutionIdCan);
+ const execItems = isExecListOpen ? executions : executions.filter((_) => _.key === selectedKey);
const nodeSettingsLayout = useDefaultLayout({ id: 'node-settings' });
@@ -506,16 +502,11 @@ export const NodeSettingsBody = ({ children, input, nodeId, output, settingsHead
aria-label='Node execution'
isOpen={isExecListOpen}
items={execItems}
- onChange={setSelectedExecKey}
+ onChange={handleSelectionChange}
onOpenChange={setIsExecListOpen}
- value={selectedExecKey}
+ value={selectedKey}
>
- {(_) => {
- let key = Ulid.construct(_.nodeExecutionId).toCanonical();
- if (key === latestExecutionIdCan) key = 'latest';
-
- return {_.name};
- }}
+ {(_) => {_.name}}
)
}
diff --git a/packages/client/src/pages/flow/nodes/wait.tsx b/packages/client/src/pages/flow/nodes/wait.tsx
new file mode 100644
index 000000000..05619ab0a
--- /dev/null
+++ b/packages/client/src/pages/flow/nodes/wait.tsx
@@ -0,0 +1,69 @@
+import { create } from '@bufbuild/protobuf';
+import { eq, useLiveQuery } from '@tanstack/react-db';
+import * as XF from '@xyflow/react';
+import { Ulid } from 'id128';
+import { use } from 'react';
+import { FiClock } from 'react-icons/fi';
+import { NodeWaitSchema } from '@the-dev-tools/spec/buf/api/flow/v1/flow_pb';
+import { NodeWaitCollectionSchema } from '@the-dev-tools/spec/tanstack-db/v1/api/flow';
+import { NumberField } from '@the-dev-tools/ui/number-field';
+import { tw } from '@the-dev-tools/ui/tailwind-literal';
+import { useApiCollection } from '~/shared/api';
+import { pick } from '~/shared/lib';
+import { FlowContext } from '../context';
+import { Handle } from '../handle';
+import { NodeSettingsBody, NodeSettingsProps, NodeTitle, SimpleNode } from '../node';
+
+const defaultNodeWait = create(NodeWaitSchema);
+
+export const WaitNode = ({ id, selected }: XF.NodeProps) => {
+ const nodeId = Ulid.fromCanonical(id).bytes;
+
+ return (
+
+
+
+
+ }
+ icon={}
+ nodeId={nodeId}
+ selected={selected}
+ >
+ Wait
+
+ );
+};
+
+export const WaitSettings = ({ nodeId }: NodeSettingsProps) => {
+ const collection = useApiCollection(NodeWaitCollectionSchema);
+
+ const data =
+ useLiveQuery(
+ (_) =>
+ _.from({ item: collection })
+ .where((_) => eq(_.item.nodeId, nodeId))
+ .select((_) => pick(_.item, 'durationMs'))
+ .findOne(),
+ [collection, nodeId],
+ ).data ?? defaultNodeWait;
+
+ const { isReadOnly = false } = use(FlowContext);
+
+ return (
+
+
+ collection.utils.updatePaced({ durationMs: BigInt(_), nodeId })}
+ value={Number(data.durationMs)}
+ />
+
+
+ );
+};
diff --git a/packages/client/src/pages/flow/nodes/ws-connection.tsx b/packages/client/src/pages/flow/nodes/ws-connection.tsx
new file mode 100644
index 000000000..89ca512f4
--- /dev/null
+++ b/packages/client/src/pages/flow/nodes/ws-connection.tsx
@@ -0,0 +1,144 @@
+import { create } from '@bufbuild/protobuf';
+import { eq, useLiveQuery } from '@tanstack/react-db';
+import { useRouter } from '@tanstack/react-router';
+import * as XF from '@xyflow/react';
+import { Ulid } from 'id128';
+import { use } from 'react';
+import { FiExternalLink, FiWifi } from 'react-icons/fi';
+import { HandleKind, NodeWsConnectionSchema } from '@the-dev-tools/spec/buf/api/flow/v1/flow_pb';
+import { NodeWsConnectionCollectionSchema } from '@the-dev-tools/spec/tanstack-db/v1/api/flow';
+import { WebSocketCollectionSchema } from '@the-dev-tools/spec/tanstack-db/v1/api/web_socket';
+import { ButtonAsLink } from '@the-dev-tools/ui/button';
+import { FieldLabel } from '@the-dev-tools/ui/field';
+import { tw } from '@the-dev-tools/ui/tailwind-literal';
+import { ReferenceContext, ReferenceField } from '~/features/expression';
+import { WebSocketHeaderTable } from '~/pages/websocket/@x/flow';
+import { useApiCollection } from '~/shared/api';
+import { eqStruct, pick } from '~/shared/lib';
+import { routes } from '~/shared/routes';
+import { FlowContext } from '../context';
+import { Handle } from '../handle';
+import { NodeSettingsBody, NodeSettingsProps, SimpleNode } from '../node';
+
+const defaultNodeWsConnection = create(NodeWsConnectionSchema);
+
+export const WsConnectionNode = ({ id, selected }: XF.NodeProps) => {
+ const nodeId = Ulid.fromCanonical(id).bytes;
+
+ const nodeWsCollection = useApiCollection(NodeWsConnectionCollectionSchema);
+ const wsCollection = useApiCollection(WebSocketCollectionSchema);
+
+ const { websocketId } =
+ useLiveQuery(
+ (_) =>
+ _.from({ item: nodeWsCollection })
+ .where((_) => eq(_.item.nodeId, nodeId))
+ .select((_) => pick(_.item, 'websocketId'))
+ .findOne(),
+ [nodeWsCollection, nodeId],
+ ).data ?? defaultNodeWsConnection;
+
+ const { url } =
+ useLiveQuery(
+ (_) =>
+ _.from({ item: wsCollection })
+ .where((_) => (websocketId ? eqStruct({ websocketId })(_) : eq(true, false)))
+ .select((_) => pick(_.item, 'url'))
+ .findOne(),
+ [wsCollection, websocketId],
+ ).data ?? {};
+
+ return (
+
+
+
+
+
+ }
+ icon={}
+ nodeId={nodeId}
+ selected={selected}
+ title='WS Connection'
+ >
+
+
WS
+
{url ?? 'No URL'}
+
+ {!isReadOnly && (
+
+
+
+ )}
+
+
void setValue(_)}
+ placeholder='Enter your GraphQL query...'
+ readOnly={isReadOnly}
+ theme={theme}
+ value={value ?? ''}
+ />
+
@@ -59,9 +70,11 @@ export const GraphQLTopBar = ({ deltaGraphqlId, graphqlId }: GraphQLTopBarProps)
onContextMenu={onContextMenu}
onPress={() => void edit()}
>
- {item?.name}
+ {name}
)}
+
+
@@ -80,7 +93,13 @@ export const GraphQLTopBar = ({ deltaGraphqlId, graphqlId }: GraphQLTopBarProps)
@@ -88,34 +107,24 @@ export const GraphQLTopBar = ({ deltaGraphqlId, graphqlId }: GraphQLTopBarProps)