mirror of
https://github.com/langflow-ai/langflow.git
synced 2026-07-24 00:39:35 +08:00
feat(collab): add frontend collaboration editing and edge-case hardening
Introduce an opt-in beta collaboration mode on the flow canvas that emits local graph operation batches over the collab WebSocket, applies remote forward_ops without full-flow autosave, and shows active collaborators. Add operation diff/adapter helpers, collaboration-aware save flushing, and coverage for drag, delete, paste/import, undo/redo, metadata, and reload boundaries, including clearing stale undo history after remote ops.
This commit is contained in:
Hamza Rashid
@ -0,0 +1,127 @@
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import useFlowsManagerStore from "@/stores/flowsManagerStore";
|
||||
import type { AllNodeType, EdgeType } from "@/types/flow";
|
||||
import { applyRemoteFlowOperations } from "../flow-operation-adapter";
|
||||
|
||||
const nodeA = {
|
||||
id: "a",
|
||||
type: "genericNode",
|
||||
position: { x: 0, y: 0 },
|
||||
data: { id: "a", type: "TextInput", node: {} },
|
||||
} as AllNodeType;
|
||||
|
||||
const nodeB = {
|
||||
id: "b",
|
||||
type: "genericNode",
|
||||
position: { x: 100, y: 0 },
|
||||
data: { id: "b", type: "TextOutput", node: {} },
|
||||
} as AllNodeType;
|
||||
|
||||
const movedNodeA = {
|
||||
...nodeA,
|
||||
position: { x: 25, y: 25 },
|
||||
} as AllNodeType;
|
||||
|
||||
const edgeAb = {
|
||||
id: "e-ab",
|
||||
source: "a",
|
||||
target: "b",
|
||||
} as EdgeType;
|
||||
|
||||
function seedFlow(flowId: string, emit = jest.fn()) {
|
||||
useFlowsManagerStore.setState({
|
||||
currentFlowId: flowId,
|
||||
currentFlow: {
|
||||
id: flowId,
|
||||
name: "Flow",
|
||||
description: "",
|
||||
data: { nodes: [nodeA, nodeB], edges: [edgeAb] },
|
||||
},
|
||||
});
|
||||
useFlowStore.setState({
|
||||
nodes: [nodeA, nodeB],
|
||||
edges: [edgeAb],
|
||||
currentFlow: {
|
||||
id: flowId,
|
||||
name: "Flow",
|
||||
description: "",
|
||||
data: { nodes: [nodeA, nodeB], edges: [edgeAb] },
|
||||
},
|
||||
collaborationOperationMode: true,
|
||||
isApplyingRemoteOperations: false,
|
||||
onCollaborationOperations: emit,
|
||||
});
|
||||
return emit;
|
||||
}
|
||||
|
||||
describe("collaboration edge-case store wiring", () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it("cut emits one combined delete batch", () => {
|
||||
const emit = seedFlow("cut-flow");
|
||||
|
||||
useFlowStore.getState().setLastCopiedSelection(
|
||||
{
|
||||
nodes: [nodeA],
|
||||
edges: [],
|
||||
},
|
||||
true,
|
||||
);
|
||||
|
||||
expect(emit).toHaveBeenCalledTimes(1);
|
||||
expect(emit).toHaveBeenCalledWith([
|
||||
{ type: "delete_nodes", ids: ["a"] },
|
||||
{ type: "delete_edges", ids: ["e-ab"] },
|
||||
]);
|
||||
});
|
||||
|
||||
it("undo emits the resulting graph delta once", () => {
|
||||
const emit = seedFlow("undo-flow");
|
||||
useFlowsManagerStore.getState().takeSnapshot();
|
||||
useFlowStore.setState({
|
||||
nodes: [movedNodeA, nodeB],
|
||||
edges: [edgeAb],
|
||||
});
|
||||
|
||||
useFlowsManagerStore.getState().undo();
|
||||
|
||||
expect(emit).toHaveBeenCalledTimes(1);
|
||||
expect(emit).toHaveBeenCalledWith([
|
||||
{ type: "update_nodes", nodes: [expect.objectContaining({ id: "a" })] },
|
||||
]);
|
||||
});
|
||||
|
||||
it("redo emits the resulting graph delta once", () => {
|
||||
const emit = seedFlow("redo-flow");
|
||||
useFlowsManagerStore.getState().takeSnapshot();
|
||||
useFlowStore.setState({
|
||||
nodes: [movedNodeA, nodeB],
|
||||
edges: [edgeAb],
|
||||
});
|
||||
useFlowsManagerStore.getState().undo();
|
||||
emit.mockClear();
|
||||
|
||||
useFlowsManagerStore.getState().redo();
|
||||
|
||||
expect(emit).toHaveBeenCalledTimes(1);
|
||||
expect(emit).toHaveBeenCalledWith([
|
||||
{ type: "update_nodes", nodes: [expect.objectContaining({ id: "a" })] },
|
||||
]);
|
||||
});
|
||||
|
||||
it("remote operations clear stale undo history so undo does not resurrect remote deletes", () => {
|
||||
const emit = seedFlow("remote-delete-flow");
|
||||
useFlowsManagerStore.getState().takeSnapshot();
|
||||
|
||||
applyRemoteFlowOperations([{ type: "delete_nodes", ids: ["a"] }]);
|
||||
emit.mockClear();
|
||||
|
||||
useFlowsManagerStore.getState().undo();
|
||||
|
||||
expect(useFlowStore.getState().nodes.map((node) => node.id)).toEqual(["b"]);
|
||||
expect(useFlowStore.getState().edges).toEqual([]);
|
||||
expect(emit).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@ -0,0 +1,145 @@
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import type { AllNodeType, EdgeType } from "@/types/flow";
|
||||
|
||||
import {
|
||||
applyFlowOperationsLocally,
|
||||
applyRemoteFlowOperations,
|
||||
buildGraphDiffOperations,
|
||||
buildUpdateMetadataOperation,
|
||||
} from "../flow-operation-adapter";
|
||||
|
||||
const nodeA = {
|
||||
id: "a",
|
||||
type: "genericNode",
|
||||
position: { x: 0, y: 0 },
|
||||
data: { id: "a", type: "TextInput", node: {} },
|
||||
} as AllNodeType;
|
||||
|
||||
const nodeB = {
|
||||
id: "b",
|
||||
type: "genericNode",
|
||||
position: { x: 100, y: 0 },
|
||||
data: { id: "b", type: "TextOutput", node: {} },
|
||||
} as AllNodeType;
|
||||
|
||||
const edgeAb = {
|
||||
id: "e-ab",
|
||||
source: "a",
|
||||
target: "b",
|
||||
} as EdgeType;
|
||||
|
||||
describe("flow-operation-adapter", () => {
|
||||
beforeEach(() => {
|
||||
useFlowStore.setState({
|
||||
nodes: [nodeA, nodeB],
|
||||
edges: [edgeAb],
|
||||
currentFlow: {
|
||||
id: "flow-1",
|
||||
name: "Flow",
|
||||
description: "",
|
||||
data: { nodes: [nodeA, nodeB], edges: [edgeAb] },
|
||||
},
|
||||
collaborationOperationMode: false,
|
||||
isApplyingRemoteOperations: false,
|
||||
onCollaborationOperations: undefined,
|
||||
});
|
||||
});
|
||||
|
||||
it("buildGraphDiffOperations detects node updates and edge deletes", () => {
|
||||
const updatedA = {
|
||||
...nodeA,
|
||||
position: { x: 10, y: 10 },
|
||||
} as AllNodeType;
|
||||
|
||||
const operations = buildGraphDiffOperations(
|
||||
[nodeA, nodeB],
|
||||
[edgeAb],
|
||||
[updatedA, nodeB],
|
||||
[],
|
||||
);
|
||||
|
||||
expect(operations).toEqual(
|
||||
expect.arrayContaining([
|
||||
{ type: "update_nodes", nodes: [expect.objectContaining({ id: "a" })] },
|
||||
{ type: "delete_edges", ids: ["e-ab"] },
|
||||
]),
|
||||
);
|
||||
});
|
||||
|
||||
it("buildGraphDiffOperations represents edge reconnect as delete plus add", () => {
|
||||
const reconnectedEdge = {
|
||||
...edgeAb,
|
||||
id: "e-reconnected",
|
||||
source: "b",
|
||||
target: "a",
|
||||
} as EdgeType;
|
||||
|
||||
const operations = buildGraphDiffOperations(
|
||||
[nodeA, nodeB],
|
||||
[edgeAb],
|
||||
[nodeA, nodeB],
|
||||
[reconnectedEdge],
|
||||
);
|
||||
|
||||
expect(operations).toEqual([
|
||||
{ type: "delete_edges", ids: ["e-ab"] },
|
||||
{ type: "add_edges", edges: [reconnectedEdge] },
|
||||
]);
|
||||
});
|
||||
|
||||
it("buildUpdateMetadataOperation ignores graph collections and viewport", () => {
|
||||
const operation = buildUpdateMetadataOperation(
|
||||
{
|
||||
nodes: [nodeA],
|
||||
edges: [edgeAb],
|
||||
viewport: { x: 0, y: 0, zoom: 1 },
|
||||
theme: "old",
|
||||
removed: true,
|
||||
},
|
||||
{
|
||||
nodes: [],
|
||||
edges: [],
|
||||
viewport: { x: 10, y: 10, zoom: 2 },
|
||||
theme: "new",
|
||||
},
|
||||
);
|
||||
|
||||
expect(operation).toEqual({
|
||||
type: "update_metadata",
|
||||
fields: { theme: "new" },
|
||||
delete_keys: ["removed"],
|
||||
});
|
||||
});
|
||||
|
||||
it("applyFlowOperationsLocally applies delete_nodes with incident edges", () => {
|
||||
const result = applyFlowOperationsLocally(
|
||||
[nodeA, nodeB],
|
||||
[edgeAb],
|
||||
[{ type: "delete_nodes", ids: ["a"] }],
|
||||
);
|
||||
|
||||
expect(result.nodes.map((node) => node.id)).toEqual(["b"]);
|
||||
expect(result.edges).toEqual([]);
|
||||
});
|
||||
|
||||
it("applyRemoteFlowOperations updates the store without emitting collaboration ops", () => {
|
||||
const emit = jest.fn();
|
||||
useFlowStore.setState({
|
||||
onCollaborationOperations: emit,
|
||||
collaborationOperationMode: true,
|
||||
});
|
||||
|
||||
applyRemoteFlowOperations([
|
||||
{
|
||||
type: "update_nodes",
|
||||
nodes: [{ ...nodeA, position: { x: 25, y: 25 } } as AllNodeType],
|
||||
},
|
||||
]);
|
||||
|
||||
expect(useFlowStore.getState().nodes[0]?.position).toEqual({
|
||||
x: 25,
|
||||
y: 25,
|
||||
});
|
||||
expect(emit).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
@ -1,5 +1,7 @@
|
||||
import { renderHook } from "@testing-library/react";
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import useFlowsManagerStore from "@/stores/flowsManagerStore";
|
||||
import type { FlowType } from "@/types/flow";
|
||||
import { useDebounce } from "../../use-debounce";
|
||||
import useAutoSaveFlow from "../use-autosave-flow";
|
||||
import useSaveFlow from "../use-save-flow";
|
||||
@ -8,6 +10,7 @@ import useSaveFlow from "../use-save-flow";
|
||||
jest.mock("../use-save-flow");
|
||||
jest.mock("../../use-debounce");
|
||||
jest.mock("@/stores/flowsManagerStore");
|
||||
jest.mock("@/stores/flowStore");
|
||||
|
||||
describe("useAutoSaveFlow", () => {
|
||||
const mockSaveFlow = jest.fn();
|
||||
@ -21,6 +24,11 @@ describe("useAutoSaveFlow", () => {
|
||||
mockDebouncedFn.mockImplementation(fn);
|
||||
return mockDebouncedFn;
|
||||
});
|
||||
(useFlowStore as unknown as { getState: jest.Mock }).getState = jest.fn(
|
||||
() => ({
|
||||
collaborationOperationMode: false,
|
||||
}),
|
||||
);
|
||||
});
|
||||
|
||||
it("should return a debounced autosave function", () => {
|
||||
@ -54,12 +62,35 @@ describe("useAutoSaveFlow", () => {
|
||||
const { result } = renderHook(() => useAutoSaveFlow());
|
||||
const autoSaveFlow = result.current;
|
||||
|
||||
const mockFlow = { id: "flow-1", name: "Test Flow" } as any;
|
||||
const mockFlow = { id: "flow-1", name: "Test Flow" } as FlowType;
|
||||
autoSaveFlow(mockFlow);
|
||||
|
||||
expect(mockSaveFlow).toHaveBeenCalledWith(mockFlow);
|
||||
});
|
||||
|
||||
it("should not call saveFlow when collaboration operation mode is enabled", () => {
|
||||
(useFlowsManagerStore as unknown as jest.Mock).mockImplementation(
|
||||
(selector) => {
|
||||
const state = {
|
||||
autoSaving: true,
|
||||
autoSavingInterval: 3000,
|
||||
};
|
||||
return selector(state);
|
||||
},
|
||||
);
|
||||
(useFlowStore as unknown as jest.Mock).mockImplementation(() => undefined);
|
||||
(useFlowStore as unknown as { getState: jest.Mock }).getState = jest.fn(
|
||||
() => ({
|
||||
collaborationOperationMode: true,
|
||||
}),
|
||||
);
|
||||
|
||||
const { result } = renderHook(() => useAutoSaveFlow());
|
||||
result.current({ id: "flow-1" } as never);
|
||||
|
||||
expect(mockSaveFlow).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("should not call saveFlow when autoSaving is disabled", () => {
|
||||
(useFlowsManagerStore as unknown as jest.Mock).mockImplementation(
|
||||
(selector) => {
|
||||
@ -74,7 +105,7 @@ describe("useAutoSaveFlow", () => {
|
||||
const { result } = renderHook(() => useAutoSaveFlow());
|
||||
const autoSaveFlow = result.current;
|
||||
|
||||
const mockFlow = { id: "flow-1", name: "Test Flow" } as any;
|
||||
const mockFlow = { id: "flow-1", name: "Test Flow" } as FlowType;
|
||||
autoSaveFlow(mockFlow);
|
||||
|
||||
expect(mockSaveFlow).not.toHaveBeenCalled();
|
||||
@ -158,7 +189,7 @@ describe("useAutoSaveFlow", () => {
|
||||
);
|
||||
|
||||
const { result, rerender } = renderHook(() => useAutoSaveFlow());
|
||||
const mockFlow = { id: "flow-1", name: "Test Flow" } as any;
|
||||
const mockFlow = { id: "flow-1", name: "Test Flow" } as FlowType;
|
||||
|
||||
// AutoSaving enabled
|
||||
result.current(mockFlow);
|
||||
|
||||
@ -0,0 +1,168 @@
|
||||
import { act, renderHook, waitFor } from "@testing-library/react";
|
||||
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
|
||||
import {
|
||||
readCollaborationOperationBetaEnabled,
|
||||
writeCollaborationOperationBetaEnabled,
|
||||
} from "../collaboration-operation-beta";
|
||||
import { useFlowCollaborationEditing } from "../use-flow-collaboration-editing";
|
||||
|
||||
const mockSubmitOperations = jest.fn().mockResolvedValue({
|
||||
type: "operation.accepted",
|
||||
revision: 1,
|
||||
});
|
||||
const mockDisconnect = jest.fn();
|
||||
const mockGetFlow = jest.fn();
|
||||
const mockApplyFlowToCanvas = jest.fn();
|
||||
const mockClearUndoRedoHistory = jest.fn();
|
||||
|
||||
jest.mock("../use-flow-collaboration", () => ({
|
||||
useFlowCollaboration: jest.fn(() => ({
|
||||
status: "ready",
|
||||
connectionId: "conn-1",
|
||||
currentRevision: 0,
|
||||
users: [],
|
||||
isReady: true,
|
||||
submitOperations: mockSubmitOperations,
|
||||
disconnect: mockDisconnect,
|
||||
})),
|
||||
}));
|
||||
|
||||
jest.mock("@/controllers/API/queries/flows/use-get-flow", () => ({
|
||||
useGetFlow: () => ({ mutateAsync: mockGetFlow }),
|
||||
}));
|
||||
|
||||
jest.mock("@/hooks/flows/use-apply-flow-to-canvas", () => ({
|
||||
__esModule: true,
|
||||
default: () => mockApplyFlowToCanvas,
|
||||
}));
|
||||
|
||||
jest.mock("@/stores/flowsManagerStore", () => {
|
||||
const state = {
|
||||
currentFlowId: "flow-1",
|
||||
flows: [],
|
||||
setFlows: jest.fn(),
|
||||
setCurrentFlow: jest.fn(),
|
||||
clearUndoRedoHistory: mockClearUndoRedoHistory,
|
||||
};
|
||||
const useFlowsManagerStore = (selector?: (value: typeof state) => unknown) =>
|
||||
selector ? selector(state) : state;
|
||||
useFlowsManagerStore.getState = () => state;
|
||||
return { __esModule: true, default: useFlowsManagerStore };
|
||||
});
|
||||
|
||||
describe("useFlowCollaborationEditing", () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
mockSubmitOperations.mockResolvedValue({
|
||||
type: "operation.accepted",
|
||||
revision: 1,
|
||||
});
|
||||
writeCollaborationOperationBetaEnabled(false);
|
||||
useFlowStore.setState({
|
||||
collaborationOperationMode: false,
|
||||
onCollaborationOperations: undefined,
|
||||
flushCollaborationSave: undefined,
|
||||
nodes: [],
|
||||
edges: [],
|
||||
currentFlow: {
|
||||
id: "flow-1",
|
||||
name: "Flow",
|
||||
description: "",
|
||||
data: { nodes: [], edges: [] },
|
||||
},
|
||||
});
|
||||
mockGetFlow.mockResolvedValue({
|
||||
id: "flow-1",
|
||||
name: "Flow",
|
||||
description: "",
|
||||
data: { nodes: [], edges: [] },
|
||||
});
|
||||
});
|
||||
|
||||
it("does not enable collaboration mode when the beta toggle is off", async () => {
|
||||
const { useFlowCollaboration } = jest.requireMock(
|
||||
"../use-flow-collaboration",
|
||||
);
|
||||
|
||||
renderHook(() =>
|
||||
useFlowCollaborationEditing({
|
||||
flowId: "flow-1",
|
||||
}),
|
||||
);
|
||||
|
||||
await act(async () => {});
|
||||
|
||||
expect(useFlowCollaboration).toHaveBeenCalledWith(
|
||||
expect.objectContaining({ enabled: false }),
|
||||
);
|
||||
expect(useFlowStore.getState().collaborationOperationMode).toBe(false);
|
||||
});
|
||||
|
||||
it("enables collaboration mode and reloads when the beta toggle is turned on", async () => {
|
||||
const { result } = renderHook(() =>
|
||||
useFlowCollaborationEditing({
|
||||
flowId: "flow-1",
|
||||
}),
|
||||
);
|
||||
|
||||
await act(async () => {
|
||||
await result.current.setBetaEnabled(true);
|
||||
});
|
||||
|
||||
expect(readCollaborationOperationBetaEnabled()).toBe(true);
|
||||
expect(mockGetFlow).toHaveBeenCalledWith({ id: "flow-1" });
|
||||
expect(mockApplyFlowToCanvas).toHaveBeenCalled();
|
||||
expect(useFlowStore.getState().collaborationOperationMode).toBe(true);
|
||||
});
|
||||
|
||||
it("submits operations emitted from the flow store when collaboration mode is active", async () => {
|
||||
writeCollaborationOperationBetaEnabled(true);
|
||||
|
||||
renderHook(() =>
|
||||
useFlowCollaborationEditing({
|
||||
flowId: "flow-1",
|
||||
}),
|
||||
);
|
||||
|
||||
await act(async () => {});
|
||||
|
||||
await act(async () => {
|
||||
useFlowStore
|
||||
.getState()
|
||||
.onCollaborationOperations?.([{ type: "delete_edges", ids: ["e1"] }]);
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockSubmitOperations).toHaveBeenCalledWith([
|
||||
{ type: "delete_edges", ids: ["e1"] },
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
it("reloads and drops pending operations when submit fails", async () => {
|
||||
mockSubmitOperations.mockRejectedValueOnce(
|
||||
new Error("Stale base revision"),
|
||||
);
|
||||
writeCollaborationOperationBetaEnabled(true);
|
||||
|
||||
renderHook(() =>
|
||||
useFlowCollaborationEditing({
|
||||
flowId: "flow-1",
|
||||
}),
|
||||
);
|
||||
|
||||
await act(async () => {});
|
||||
|
||||
await act(async () => {
|
||||
useFlowStore
|
||||
.getState()
|
||||
.onCollaborationOperations?.([{ type: "delete_edges", ids: ["e1"] }]);
|
||||
});
|
||||
|
||||
await waitFor(() => {
|
||||
expect(mockGetFlow).toHaveBeenCalledWith({ id: "flow-1" });
|
||||
});
|
||||
});
|
||||
});
|
||||
@ -154,6 +154,12 @@ describe("useFlowCollaboration", () => {
|
||||
expect(instances).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("should not open a socket when enabled is false", async () => {
|
||||
await mountHook({ flowId: "flow-1", enabled: false });
|
||||
|
||||
expect(instances).toHaveLength(0);
|
||||
});
|
||||
|
||||
it("should open the socket, send session.start, and become ready on session.ready", async () => {
|
||||
const { result } = await mountHook({ flowId: "flow-1" });
|
||||
|
||||
|
||||
@ -1,4 +1,6 @@
|
||||
import { renderHook } from "@testing-library/react";
|
||||
import type { FlowType } from "@/types/flow";
|
||||
import type { FlowOperation } from "@/types/flow-operations";
|
||||
import useSaveFlow from "../use-save-flow";
|
||||
|
||||
const mockSetFlows = jest.fn();
|
||||
@ -8,8 +10,31 @@ const mockSetCurrentFlow = jest.fn();
|
||||
const mockGetFlow = jest.fn();
|
||||
const mockMutate = jest.fn();
|
||||
|
||||
let flowStoreState: any;
|
||||
let flowsManagerState: any;
|
||||
type MockFlowStoreState = {
|
||||
collaborationOperationMode: boolean;
|
||||
flushCollaborationSave?: () => Promise<void>;
|
||||
onCollaborationOperations?: (operations: FlowOperation[]) => void;
|
||||
currentFlow: FlowType | null;
|
||||
nodes: FlowType["data"]["nodes"];
|
||||
edges: FlowType["data"]["edges"];
|
||||
reactFlowInstance: {
|
||||
getViewport: jest.Mock;
|
||||
} | null;
|
||||
onFlowPage: boolean;
|
||||
setCurrentFlow: typeof mockSetCurrentFlow;
|
||||
};
|
||||
|
||||
type MockFlowsManagerState = {
|
||||
currentFlow: FlowType | null;
|
||||
flows: FlowType[];
|
||||
setFlows: typeof mockSetFlows;
|
||||
setSaveLoading: typeof mockSetSaveLoading;
|
||||
};
|
||||
|
||||
type StoreSelector<T> = (state: T) => unknown;
|
||||
|
||||
let flowStoreState: MockFlowStoreState;
|
||||
let flowsManagerState: MockFlowsManagerState;
|
||||
|
||||
jest.mock("@/controllers/API/queries/flows/use-get-flow", () => ({
|
||||
useGetFlow: () => ({ mutate: mockGetFlow }),
|
||||
@ -21,14 +46,16 @@ jest.mock("@/controllers/API/queries/flows/use-patch-update-flow", () => ({
|
||||
|
||||
jest.mock("@/stores/alertStore", () => ({
|
||||
__esModule: true,
|
||||
default: (selector: any) =>
|
||||
default: (
|
||||
selector: StoreSelector<{ setErrorData: typeof mockSetErrorData }>,
|
||||
) =>
|
||||
selector({
|
||||
setErrorData: mockSetErrorData,
|
||||
}),
|
||||
}));
|
||||
|
||||
jest.mock("@/stores/flowStore", () => {
|
||||
const useFlowStore = (selector: any) =>
|
||||
const useFlowStore = (selector?: StoreSelector<MockFlowStoreState>) =>
|
||||
selector ? selector(flowStoreState) : flowStoreState;
|
||||
useFlowStore.getState = () => flowStoreState;
|
||||
|
||||
@ -39,8 +66,9 @@ jest.mock("@/stores/flowStore", () => {
|
||||
});
|
||||
|
||||
jest.mock("@/stores/flowsManagerStore", () => {
|
||||
const useFlowsManagerStore = (selector: any) =>
|
||||
selector ? selector(flowsManagerState) : flowsManagerState;
|
||||
const useFlowsManagerStore = (
|
||||
selector?: StoreSelector<MockFlowsManagerState>,
|
||||
) => (selector ? selector(flowsManagerState) : flowsManagerState);
|
||||
useFlowsManagerStore.getState = () => flowsManagerState;
|
||||
|
||||
return {
|
||||
@ -65,9 +93,11 @@ describe("useSaveFlow", () => {
|
||||
folder_id: "folder-1",
|
||||
endpoint_name: "saved-flow",
|
||||
locked: false,
|
||||
};
|
||||
} as FlowType;
|
||||
|
||||
flowStoreState = {
|
||||
collaborationOperationMode: false,
|
||||
flushCollaborationSave: undefined,
|
||||
currentFlow: {
|
||||
...savedFlow,
|
||||
data: {
|
||||
@ -104,6 +134,68 @@ describe("useSaveFlow", () => {
|
||||
});
|
||||
});
|
||||
|
||||
it("flushes collaboration operations instead of PATCH when collaboration mode is active", async () => {
|
||||
const flushCollaborationSave = jest.fn().mockResolvedValue(undefined);
|
||||
flowStoreState = {
|
||||
...flowStoreState,
|
||||
collaborationOperationMode: true,
|
||||
flushCollaborationSave,
|
||||
};
|
||||
|
||||
const { result } = renderHook(() => useSaveFlow());
|
||||
|
||||
await expect(result.current()).resolves.toBeUndefined();
|
||||
|
||||
expect(flushCollaborationSave).toHaveBeenCalledTimes(1);
|
||||
expect(mockMutate).not.toHaveBeenCalled();
|
||||
expect(mockSetSaveLoading).toHaveBeenCalledWith(true);
|
||||
expect(mockSetSaveLoading).toHaveBeenCalledWith(false);
|
||||
});
|
||||
|
||||
it("emits flow.data metadata updates before flushing collaboration save", async () => {
|
||||
const flushCollaborationSave = jest.fn().mockResolvedValue(undefined);
|
||||
const onCollaborationOperations = jest.fn();
|
||||
flowsManagerState.currentFlow = {
|
||||
...flowsManagerState.currentFlow,
|
||||
data: {
|
||||
nodes: [],
|
||||
edges: [],
|
||||
viewport: { x: 0, y: 0, zoom: 1 },
|
||||
theme: "old",
|
||||
stale_key: true,
|
||||
},
|
||||
} as FlowType;
|
||||
flowStoreState = {
|
||||
...flowStoreState,
|
||||
collaborationOperationMode: true,
|
||||
flushCollaborationSave,
|
||||
onCollaborationOperations,
|
||||
currentFlow: {
|
||||
...flowStoreState.currentFlow,
|
||||
data: {
|
||||
nodes: [],
|
||||
edges: [],
|
||||
viewport: { x: 10, y: 10, zoom: 2 },
|
||||
theme: "new",
|
||||
},
|
||||
} as FlowType,
|
||||
};
|
||||
|
||||
const { result } = renderHook(() => useSaveFlow());
|
||||
|
||||
await expect(result.current()).resolves.toBeUndefined();
|
||||
|
||||
expect(onCollaborationOperations).toHaveBeenCalledWith([
|
||||
{
|
||||
type: "update_metadata",
|
||||
fields: { theme: "new" },
|
||||
delete_keys: ["stale_key"],
|
||||
},
|
||||
]);
|
||||
expect(flushCollaborationSave).toHaveBeenCalledTimes(1);
|
||||
expect(mockMutate).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("persists empty-node flows instead of leaving the save promise pending", async () => {
|
||||
const { result } = renderHook(() => useSaveFlow());
|
||||
|
||||
@ -142,7 +234,7 @@ describe("useSaveFlow", () => {
|
||||
endpoint_name: "saved-flow",
|
||||
locked: false,
|
||||
is_component: false,
|
||||
};
|
||||
} as FlowType;
|
||||
|
||||
flowStoreState = {
|
||||
currentFlow: null,
|
||||
@ -151,6 +243,7 @@ describe("useSaveFlow", () => {
|
||||
reactFlowInstance: null,
|
||||
onFlowPage: false,
|
||||
setCurrentFlow: mockSetCurrentFlow,
|
||||
collaborationOperationMode: false,
|
||||
};
|
||||
|
||||
flowsManagerState = {
|
||||
|
||||
@ -0,0 +1,62 @@
|
||||
import { renderHook } from "@testing-library/react";
|
||||
|
||||
import useUploadFlow from "../use-upload-flow";
|
||||
|
||||
const mockGetObjectsFromFilelist = jest.fn();
|
||||
const mockProcessDataFromFlow = jest.fn();
|
||||
const mockAddFlow = jest.fn();
|
||||
const mockPaste = jest.fn();
|
||||
|
||||
jest.mock("@/helpers/get-objects-from-filelist", () => ({
|
||||
getObjectsFromFilelist: (...args: unknown[]) =>
|
||||
mockGetObjectsFromFilelist(...args),
|
||||
}));
|
||||
|
||||
jest.mock("@/utils/reactflowUtils", () => ({
|
||||
processDataFromFlow: (...args: unknown[]) => mockProcessDataFromFlow(...args),
|
||||
}));
|
||||
|
||||
jest.mock("../use-add-flow", () => ({
|
||||
__esModule: true,
|
||||
default: () => mockAddFlow,
|
||||
}));
|
||||
|
||||
jest.mock("@/stores/flowStore", () => {
|
||||
const useFlowStoreMock = (
|
||||
selector: (state: { paste: jest.Mock }) => unknown,
|
||||
) => selector({ paste: mockPaste });
|
||||
return {
|
||||
__esModule: true,
|
||||
default: useFlowStoreMock,
|
||||
};
|
||||
});
|
||||
|
||||
describe("useUploadFlow", () => {
|
||||
beforeEach(() => {
|
||||
jest.clearAllMocks();
|
||||
});
|
||||
|
||||
it("uses paste for canvas JSON drop/import so collaboration mode can emit add operations", async () => {
|
||||
const flow = {
|
||||
id: "flow-1",
|
||||
name: "Imported Flow",
|
||||
data: {
|
||||
nodes: [{ id: "n1" }],
|
||||
edges: [{ id: "e1", source: "n1", target: "n2" }],
|
||||
},
|
||||
};
|
||||
mockGetObjectsFromFilelist.mockResolvedValue([flow]);
|
||||
mockProcessDataFromFlow.mockResolvedValue(undefined);
|
||||
const { result } = renderHook(() => useUploadFlow());
|
||||
const position = { x: 10, y: 20 };
|
||||
|
||||
await result.current({
|
||||
files: [new File(["{}"], "flow.json", { type: "application/json" })],
|
||||
position,
|
||||
});
|
||||
|
||||
expect(mockProcessDataFromFlow).toHaveBeenCalledWith(flow);
|
||||
expect(mockPaste).toHaveBeenCalledWith(flow.data, position);
|
||||
expect(mockAddFlow).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
22
src/frontend/src/hooks/flows/collaboration-operation-beta.ts
Normal file
22
src/frontend/src/hooks/flows/collaboration-operation-beta.ts
Normal file
@ -0,0 +1,22 @@
|
||||
const COLLABORATION_OPERATION_BETA_STORAGE_KEY =
|
||||
"langflow_collaboration_operation_beta";
|
||||
|
||||
export function readCollaborationOperationBetaEnabled(): boolean {
|
||||
if (typeof window === "undefined") {
|
||||
return false;
|
||||
}
|
||||
return (
|
||||
window.localStorage.getItem(COLLABORATION_OPERATION_BETA_STORAGE_KEY) ===
|
||||
"true"
|
||||
);
|
||||
}
|
||||
|
||||
export function writeCollaborationOperationBetaEnabled(enabled: boolean): void {
|
||||
if (typeof window === "undefined") {
|
||||
return;
|
||||
}
|
||||
window.localStorage.setItem(
|
||||
COLLABORATION_OPERATION_BETA_STORAGE_KEY,
|
||||
enabled ? "true" : "false",
|
||||
);
|
||||
}
|
||||
199
src/frontend/src/hooks/flows/flow-operation-adapter.ts
Normal file
199
src/frontend/src/hooks/flows/flow-operation-adapter.ts
Normal file
@ -0,0 +1,199 @@
|
||||
import { cloneDeep } from "lodash";
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import useFlowsManagerStore from "@/stores/flowsManagerStore";
|
||||
import type { AllNodeType, EdgeType, FlowType } from "@/types/flow";
|
||||
import type { FlowOperation } from "@/types/flow-operations";
|
||||
import { cleanEdges } from "@/utils/reactflowUtils";
|
||||
import { getInputsAndOutputs } from "@/utils/storeUtils";
|
||||
import { coalesceDeleteIds } from "./flow-operation-diff";
|
||||
|
||||
export type { FlowMutationOptions } from "@/types/flow-operations";
|
||||
export {
|
||||
buildGraphDiffOperations,
|
||||
buildUpdateMetadataOperation,
|
||||
buildUpdateNodesOperation,
|
||||
} from "./flow-operation-diff";
|
||||
|
||||
function applyDeleteNodes(
|
||||
nodes: AllNodeType[],
|
||||
edges: EdgeType[],
|
||||
ids: string[],
|
||||
): { nodes: AllNodeType[]; edges: EdgeType[] } {
|
||||
const nodeIds = new Set(coalesceDeleteIds(ids));
|
||||
const nextNodes = nodes.filter((node) => !nodeIds.has(node.id));
|
||||
const nextEdges = edges.filter(
|
||||
(edge) => !nodeIds.has(edge.source) && !nodeIds.has(edge.target),
|
||||
);
|
||||
return { nodes: nextNodes, edges: nextEdges };
|
||||
}
|
||||
|
||||
function applyDeleteEdges(edges: EdgeType[], ids: string[]): EdgeType[] {
|
||||
const edgeIds = new Set(coalesceDeleteIds(ids));
|
||||
return edges.filter((edge) => !edgeIds.has(edge.id));
|
||||
}
|
||||
|
||||
export function applyFlowOperationsLocally(
|
||||
nodes: AllNodeType[],
|
||||
edges: EdgeType[],
|
||||
operations: FlowOperation[],
|
||||
): { nodes: AllNodeType[]; edges: EdgeType[] } {
|
||||
let nextNodes = cloneDeep(nodes);
|
||||
let nextEdges = cloneDeep(edges);
|
||||
|
||||
for (const operation of operations) {
|
||||
switch (operation.type) {
|
||||
case "add_nodes": {
|
||||
const existingIds = new Set(nextNodes.map((node) => node.id));
|
||||
for (const node of operation.nodes) {
|
||||
if (!existingIds.has(node.id)) {
|
||||
nextNodes.push(cloneDeep(node));
|
||||
existingIds.add(node.id);
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "update_nodes": {
|
||||
const nodeMap = new Map(nextNodes.map((node) => [node.id, node]));
|
||||
for (const node of operation.nodes) {
|
||||
if (nodeMap.has(node.id)) {
|
||||
nodeMap.set(node.id, cloneDeep(node));
|
||||
}
|
||||
}
|
||||
nextNodes = Array.from(nodeMap.values());
|
||||
break;
|
||||
}
|
||||
case "delete_nodes": {
|
||||
const deleted = applyDeleteNodes(nextNodes, nextEdges, operation.ids);
|
||||
nextNodes = deleted.nodes;
|
||||
nextEdges = deleted.edges;
|
||||
break;
|
||||
}
|
||||
case "add_edges": {
|
||||
const existingIds = new Set(nextEdges.map((edge) => edge.id));
|
||||
for (const edge of operation.edges) {
|
||||
if (!existingIds.has(edge.id)) {
|
||||
nextEdges.push(cloneDeep(edge));
|
||||
existingIds.add(edge.id);
|
||||
}
|
||||
}
|
||||
break;
|
||||
}
|
||||
case "delete_edges": {
|
||||
nextEdges = applyDeleteEdges(nextEdges, operation.ids);
|
||||
break;
|
||||
}
|
||||
case "update_metadata":
|
||||
break;
|
||||
default:
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
const cleaned = cleanEdges(nextNodes, nextEdges);
|
||||
return { nodes: nextNodes, edges: cleaned.edges };
|
||||
}
|
||||
|
||||
export function applyRemoteFlowOperations(operations: FlowOperation[]): void {
|
||||
const store = useFlowStore.getState();
|
||||
const metadataOps = operations.filter(
|
||||
(operation) => operation.type === "update_metadata",
|
||||
);
|
||||
const graphOps = operations.filter(
|
||||
(operation) => operation.type !== "update_metadata",
|
||||
);
|
||||
|
||||
const { nodes, edges } = applyFlowOperationsLocally(
|
||||
store.nodes,
|
||||
store.edges,
|
||||
graphOps,
|
||||
);
|
||||
|
||||
const { inputs, outputs } = getInputsAndOutputs(nodes);
|
||||
useFlowStore.setState({ isApplyingRemoteOperations: true });
|
||||
try {
|
||||
useFlowStore.setState({
|
||||
nodes,
|
||||
edges,
|
||||
flowState: undefined,
|
||||
inputs,
|
||||
outputs,
|
||||
hasIO: inputs.length > 0 || outputs.length > 0,
|
||||
});
|
||||
useFlowStore.getState().updateCurrentFlow({ nodes, edges });
|
||||
|
||||
const currentFlow = useFlowStore.getState().currentFlow;
|
||||
if (metadataOps.length > 0 && currentFlow?.data) {
|
||||
const nextData = { ...currentFlow.data };
|
||||
for (const operation of metadataOps) {
|
||||
if (operation.type !== "update_metadata") {
|
||||
continue;
|
||||
}
|
||||
for (const [key, value] of Object.entries(operation.fields)) {
|
||||
if (key === "nodes" || key === "edges") {
|
||||
continue;
|
||||
}
|
||||
nextData[key] = value;
|
||||
}
|
||||
for (const key of operation.delete_keys ?? []) {
|
||||
if (key === "nodes" || key === "edges") {
|
||||
continue;
|
||||
}
|
||||
delete nextData[key];
|
||||
}
|
||||
}
|
||||
useFlowStore.getState().setCurrentFlow({
|
||||
...currentFlow,
|
||||
data: {
|
||||
...nextData,
|
||||
nodes,
|
||||
edges,
|
||||
},
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
useFlowStore.setState({ isApplyingRemoteOperations: false });
|
||||
}
|
||||
useFlowsManagerStore.getState().clearUndoRedoHistory?.();
|
||||
}
|
||||
|
||||
export function syncSavedFlowStateFromCanvas(): void {
|
||||
const flowStore = useFlowStore.getState();
|
||||
const currentFlow = flowStore.currentFlow;
|
||||
if (!currentFlow) {
|
||||
return;
|
||||
}
|
||||
|
||||
const viewport = flowStore.reactFlowInstance?.getViewport() ??
|
||||
currentFlow.data?.viewport ?? {
|
||||
x: 0,
|
||||
y: 0,
|
||||
zoom: 1,
|
||||
};
|
||||
|
||||
const updatedFlow: FlowType = {
|
||||
...currentFlow,
|
||||
data: {
|
||||
...currentFlow.data,
|
||||
nodes: flowStore.nodes,
|
||||
edges: flowStore.edges,
|
||||
viewport,
|
||||
},
|
||||
};
|
||||
|
||||
flowStore.setCurrentFlow(updatedFlow);
|
||||
|
||||
const flows = useFlowsManagerStore.getState().flows;
|
||||
if (!flows) {
|
||||
return;
|
||||
}
|
||||
|
||||
useFlowsManagerStore
|
||||
.getState()
|
||||
.setFlows(
|
||||
flows.map((flow) => (flow.id === updatedFlow.id ? updatedFlow : flow)),
|
||||
);
|
||||
|
||||
if (flowStore.onFlowPage) {
|
||||
useFlowsManagerStore.getState().setCurrentFlow(updatedFlow);
|
||||
}
|
||||
}
|
||||
161
src/frontend/src/hooks/flows/flow-operation-diff.ts
Normal file
161
src/frontend/src/hooks/flows/flow-operation-diff.ts
Normal file
@ -0,0 +1,161 @@
|
||||
import { cloneDeep, isEqual } from "lodash";
|
||||
import type { AllNodeType, EdgeType } from "@/types/flow";
|
||||
import type { FlowOperation, UpdateMetadataOp } from "@/types/flow-operations";
|
||||
|
||||
export function coalesceDeleteIds(ids: string[]): string[] {
|
||||
const seen = new Set<string>();
|
||||
const result: string[] = [];
|
||||
for (const id of ids) {
|
||||
if (!seen.has(id)) {
|
||||
seen.add(id);
|
||||
result.push(id);
|
||||
}
|
||||
}
|
||||
return result;
|
||||
}
|
||||
|
||||
function nodeSnapshot(node: AllNodeType): AllNodeType {
|
||||
const { selected: _selected, ...rest } = node;
|
||||
return rest as AllNodeType;
|
||||
}
|
||||
|
||||
function edgeSnapshot(edge: EdgeType): EdgeType {
|
||||
const { selected: _selected, ...rest } = edge;
|
||||
return rest as EdgeType;
|
||||
}
|
||||
|
||||
function nodesEqual(a: AllNodeType, b: AllNodeType): boolean {
|
||||
return isEqual(nodeSnapshot(a), nodeSnapshot(b));
|
||||
}
|
||||
|
||||
function edgesEqual(a: EdgeType, b: EdgeType): boolean {
|
||||
return isEqual(edgeSnapshot(a), edgeSnapshot(b));
|
||||
}
|
||||
|
||||
export function buildGraphDiffOperations(
|
||||
prevNodes: AllNodeType[],
|
||||
prevEdges: EdgeType[],
|
||||
nextNodes: AllNodeType[],
|
||||
nextEdges: EdgeType[],
|
||||
): FlowOperation[] {
|
||||
const operations: FlowOperation[] = [];
|
||||
|
||||
const prevNodeMap = new Map(prevNodes.map((node) => [node.id, node]));
|
||||
const nextNodeMap = new Map(nextNodes.map((node) => [node.id, node]));
|
||||
|
||||
const addedNodes: AllNodeType[] = [];
|
||||
const updatedNodes: AllNodeType[] = [];
|
||||
const deletedNodeIds: string[] = [];
|
||||
|
||||
for (const node of nextNodes) {
|
||||
const previous = prevNodeMap.get(node.id);
|
||||
if (!previous) {
|
||||
addedNodes.push(cloneDeep(node));
|
||||
continue;
|
||||
}
|
||||
if (!nodesEqual(previous, node)) {
|
||||
updatedNodes.push(cloneDeep(node));
|
||||
}
|
||||
}
|
||||
|
||||
for (const node of prevNodes) {
|
||||
if (!nextNodeMap.has(node.id)) {
|
||||
deletedNodeIds.push(node.id);
|
||||
}
|
||||
}
|
||||
|
||||
if (addedNodes.length > 0) {
|
||||
operations.push({ type: "add_nodes", nodes: addedNodes });
|
||||
}
|
||||
if (updatedNodes.length > 0) {
|
||||
operations.push({ type: "update_nodes", nodes: updatedNodes });
|
||||
}
|
||||
if (deletedNodeIds.length > 0) {
|
||||
operations.push({
|
||||
type: "delete_nodes",
|
||||
ids: coalesceDeleteIds(deletedNodeIds),
|
||||
});
|
||||
}
|
||||
|
||||
const prevEdgeMap = new Map(prevEdges.map((edge) => [edge.id, edge]));
|
||||
const nextEdgeMap = new Map(nextEdges.map((edge) => [edge.id, edge]));
|
||||
|
||||
const addedEdges: EdgeType[] = [];
|
||||
const deletedEdgeIds: string[] = [];
|
||||
|
||||
for (const edge of nextEdges) {
|
||||
const previous = prevEdgeMap.get(edge.id);
|
||||
if (!previous) {
|
||||
addedEdges.push(cloneDeep(edge));
|
||||
continue;
|
||||
}
|
||||
if (!edgesEqual(previous, edge)) {
|
||||
deletedEdgeIds.push(edge.id);
|
||||
addedEdges.push(cloneDeep(edge));
|
||||
}
|
||||
}
|
||||
|
||||
for (const edge of prevEdges) {
|
||||
if (!nextEdgeMap.has(edge.id)) {
|
||||
deletedEdgeIds.push(edge.id);
|
||||
}
|
||||
}
|
||||
|
||||
if (deletedEdgeIds.length > 0) {
|
||||
operations.push({
|
||||
type: "delete_edges",
|
||||
ids: coalesceDeleteIds(deletedEdgeIds),
|
||||
});
|
||||
}
|
||||
if (addedEdges.length > 0) {
|
||||
operations.push({ type: "add_edges", edges: addedEdges });
|
||||
}
|
||||
|
||||
return operations;
|
||||
}
|
||||
|
||||
export function buildUpdateNodesOperation(nodes: AllNodeType[]): FlowOperation {
|
||||
return {
|
||||
type: "update_nodes",
|
||||
nodes: nodes.map((node) => cloneDeep(node)),
|
||||
};
|
||||
}
|
||||
|
||||
export function buildUpdateMetadataOperation(
|
||||
previousData: Record<string, unknown> | null | undefined,
|
||||
nextData: Record<string, unknown> | null | undefined,
|
||||
): UpdateMetadataOp | null {
|
||||
const previous = previousData ?? {};
|
||||
const next = nextData ?? {};
|
||||
const fields: Record<string, unknown> = {};
|
||||
const deleteKeys: string[] = [];
|
||||
const ignoredKeys = new Set(["nodes", "edges", "viewport"]);
|
||||
|
||||
for (const [key, value] of Object.entries(next)) {
|
||||
if (ignoredKeys.has(key)) {
|
||||
continue;
|
||||
}
|
||||
if (!isEqual(previous[key], value)) {
|
||||
fields[key] = cloneDeep(value);
|
||||
}
|
||||
}
|
||||
|
||||
for (const key of Object.keys(previous)) {
|
||||
if (ignoredKeys.has(key)) {
|
||||
continue;
|
||||
}
|
||||
if (!(key in next)) {
|
||||
deleteKeys.push(key);
|
||||
}
|
||||
}
|
||||
|
||||
if (Object.keys(fields).length === 0 && deleteKeys.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
return {
|
||||
type: "update_metadata",
|
||||
fields,
|
||||
delete_keys: coalesceDeleteIds(deleteKeys),
|
||||
};
|
||||
}
|
||||
@ -1,3 +1,4 @@
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import useFlowsManagerStore from "@/stores/flowsManagerStore";
|
||||
import type { FlowType } from "@/types/flow";
|
||||
import { useDebounce } from "../use-debounce";
|
||||
@ -11,7 +12,7 @@ const useAutoSaveFlow = () => {
|
||||
);
|
||||
|
||||
const autoSaveFlow = useDebounce((flow?: FlowType) => {
|
||||
if (autoSaving) {
|
||||
if (autoSaving && !useFlowStore.getState().collaborationOperationMode) {
|
||||
saveFlow(flow);
|
||||
}
|
||||
}, autoSavingInterval);
|
||||
|
||||
182
src/frontend/src/hooks/flows/use-flow-collaboration-editing.ts
Normal file
182
src/frontend/src/hooks/flows/use-flow-collaboration-editing.ts
Normal file
@ -0,0 +1,182 @@
|
||||
import { useCallback, useEffect, useRef, useState } from "react";
|
||||
import { useGetFlow } from "@/controllers/API/queries/flows/use-get-flow";
|
||||
import {
|
||||
readCollaborationOperationBetaEnabled,
|
||||
writeCollaborationOperationBetaEnabled,
|
||||
} from "@/hooks/flows/collaboration-operation-beta";
|
||||
import {
|
||||
applyRemoteFlowOperations,
|
||||
syncSavedFlowStateFromCanvas,
|
||||
} from "@/hooks/flows/flow-operation-adapter";
|
||||
import useApplyFlowToCanvas from "@/hooks/flows/use-apply-flow-to-canvas";
|
||||
import { useFlowCollaboration } from "@/hooks/flows/use-flow-collaboration";
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import useFlowsManagerStore from "@/stores/flowsManagerStore";
|
||||
import type { CollaborationPresenceUser } from "@/types/flow-collaboration";
|
||||
import type { FlowOperation } from "@/types/flow-operations";
|
||||
|
||||
type UseFlowCollaborationEditingOptions = {
|
||||
flowId: string | undefined;
|
||||
};
|
||||
|
||||
export type UseFlowCollaborationEditingReturn = {
|
||||
betaEnabled: boolean;
|
||||
setBetaEnabled: (enabled: boolean) => Promise<void>;
|
||||
users: CollaborationPresenceUser[];
|
||||
isCollaborationReady: boolean;
|
||||
collaborationStatus: ReturnType<typeof useFlowCollaboration>["status"];
|
||||
};
|
||||
|
||||
export function useFlowCollaborationEditing({
|
||||
flowId,
|
||||
}: UseFlowCollaborationEditingOptions): UseFlowCollaborationEditingReturn {
|
||||
const [betaEnabled, setBetaEnabledState] = useState(
|
||||
readCollaborationOperationBetaEnabled,
|
||||
);
|
||||
const applyFlowToCanvas = useApplyFlowToCanvas();
|
||||
const { mutateAsync: getFlow } = useGetFlow();
|
||||
|
||||
const submitChainRef = useRef<Promise<void>>(Promise.resolve());
|
||||
const queuedOperationsRef = useRef<FlowOperation[]>([]);
|
||||
const collaborationReadyRef = useRef(false);
|
||||
const reloadingRef = useRef(false);
|
||||
|
||||
const reloadFlowFromServer = useCallback(async () => {
|
||||
if (!flowId) {
|
||||
return;
|
||||
}
|
||||
queuedOperationsRef.current = [];
|
||||
submitChainRef.current = Promise.resolve();
|
||||
reloadingRef.current = true;
|
||||
useFlowStore.setState({ isApplyingRemoteOperations: true });
|
||||
try {
|
||||
const response = await getFlow({ id: flowId });
|
||||
if (useFlowsManagerStore.getState().currentFlowId !== flowId) {
|
||||
return;
|
||||
}
|
||||
applyFlowToCanvas(response);
|
||||
syncSavedFlowStateFromCanvas();
|
||||
useFlowsManagerStore.getState().clearUndoRedoHistory?.(flowId);
|
||||
} finally {
|
||||
reloadingRef.current = false;
|
||||
useFlowStore.setState({ isApplyingRemoteOperations: false });
|
||||
}
|
||||
}, [applyFlowToCanvas, flowId, getFlow]);
|
||||
|
||||
const submitOperationsRef = useRef<
|
||||
((operations: FlowOperation[]) => Promise<unknown>) | null
|
||||
>(null);
|
||||
|
||||
const drainQueue = useCallback(async () => {
|
||||
if (
|
||||
!collaborationReadyRef.current ||
|
||||
reloadingRef.current ||
|
||||
queuedOperationsRef.current.length === 0
|
||||
) {
|
||||
return;
|
||||
}
|
||||
|
||||
const submit = submitOperationsRef.current;
|
||||
if (!submit) {
|
||||
return;
|
||||
}
|
||||
|
||||
const batch = queuedOperationsRef.current.splice(
|
||||
0,
|
||||
queuedOperationsRef.current.length,
|
||||
);
|
||||
try {
|
||||
await submit(batch);
|
||||
} catch {
|
||||
queuedOperationsRef.current = [];
|
||||
await reloadFlowFromServer();
|
||||
}
|
||||
}, [reloadFlowFromServer]);
|
||||
|
||||
const enqueueOperations = useCallback(
|
||||
(operations: FlowOperation[]) => {
|
||||
if (!operations.length || reloadingRef.current) {
|
||||
return;
|
||||
}
|
||||
queuedOperationsRef.current.push(...operations);
|
||||
if (!collaborationReadyRef.current) {
|
||||
return;
|
||||
}
|
||||
submitChainRef.current = submitChainRef.current
|
||||
.catch(() => {})
|
||||
.then(drainQueue);
|
||||
},
|
||||
[drainQueue],
|
||||
);
|
||||
|
||||
const flushCollaborationSave = useCallback(async () => {
|
||||
await submitChainRef.current;
|
||||
if (queuedOperationsRef.current.length > 0) {
|
||||
if (!collaborationReadyRef.current) {
|
||||
throw new Error("Collaboration session is not ready");
|
||||
}
|
||||
await drainQueue();
|
||||
}
|
||||
syncSavedFlowStateFromCanvas();
|
||||
}, [drainQueue]);
|
||||
|
||||
const collaboration = useFlowCollaboration({
|
||||
flowId,
|
||||
enabled: betaEnabled && Boolean(flowId),
|
||||
onRemoteOperation: (message) => {
|
||||
applyRemoteFlowOperations(message.forward_ops);
|
||||
},
|
||||
onReloadRequired: () => {
|
||||
queuedOperationsRef.current = [];
|
||||
submitChainRef.current = Promise.resolve();
|
||||
void reloadFlowFromServer();
|
||||
},
|
||||
});
|
||||
|
||||
submitOperationsRef.current = collaboration.submitOperations;
|
||||
collaborationReadyRef.current = collaboration.isReady;
|
||||
|
||||
useEffect(() => {
|
||||
if (!collaboration.isReady || queuedOperationsRef.current.length === 0) {
|
||||
return;
|
||||
}
|
||||
submitChainRef.current = submitChainRef.current
|
||||
.catch(() => {})
|
||||
.then(drainQueue);
|
||||
}, [collaboration.isReady, drainQueue]);
|
||||
|
||||
useEffect(() => {
|
||||
useFlowStore.setState({
|
||||
collaborationOperationMode: betaEnabled,
|
||||
onCollaborationOperations: betaEnabled ? enqueueOperations : undefined,
|
||||
flushCollaborationSave: betaEnabled ? flushCollaborationSave : undefined,
|
||||
});
|
||||
|
||||
return () => {
|
||||
useFlowStore.setState({
|
||||
collaborationOperationMode: false,
|
||||
onCollaborationOperations: undefined,
|
||||
flushCollaborationSave: undefined,
|
||||
});
|
||||
};
|
||||
}, [betaEnabled, enqueueOperations, flushCollaborationSave]);
|
||||
|
||||
const setBetaEnabled = useCallback(
|
||||
async (enabled: boolean) => {
|
||||
writeCollaborationOperationBetaEnabled(enabled);
|
||||
setBetaEnabledState(enabled);
|
||||
if (flowId) {
|
||||
await reloadFlowFromServer();
|
||||
}
|
||||
},
|
||||
[flowId, reloadFlowFromServer],
|
||||
);
|
||||
|
||||
return {
|
||||
betaEnabled,
|
||||
setBetaEnabled,
|
||||
users: collaboration.users,
|
||||
isCollaborationReady: collaboration.isReady,
|
||||
collaborationStatus: collaboration.status,
|
||||
};
|
||||
}
|
||||
@ -2,6 +2,8 @@ import type { ReactFlowJsonObject } from "@xyflow/react";
|
||||
import { useTranslation } from "react-i18next";
|
||||
import { useGetFlow } from "@/controllers/API/queries/flows/use-get-flow";
|
||||
import { usePatchUpdateFlow } from "@/controllers/API/queries/flows/use-patch-update-flow";
|
||||
import { syncSavedFlowStateFromCanvas } from "@/hooks/flows/flow-operation-adapter";
|
||||
import { buildUpdateMetadataOperation } from "@/hooks/flows/flow-operation-diff";
|
||||
import useAlertStore from "@/stores/alertStore";
|
||||
import useFlowStore from "@/stores/flowStore";
|
||||
import useFlowsManagerStore from "@/stores/flowsManagerStore";
|
||||
@ -19,8 +21,42 @@ const useSaveFlow = () => {
|
||||
const { mutate } = usePatchUpdateFlow();
|
||||
|
||||
const saveFlow = async (flow?: FlowType): Promise<void> => {
|
||||
const currentFlow = useFlowStore.getState().currentFlow;
|
||||
const flowStore = useFlowStore.getState();
|
||||
const currentFlow = flowStore.currentFlow;
|
||||
const currentSavedFlow = useFlowsManagerStore.getState().currentFlow;
|
||||
|
||||
if (flowStore.collaborationOperationMode) {
|
||||
setSaveLoading(true);
|
||||
try {
|
||||
const flowToSave = flow ?? currentFlow;
|
||||
const metadataOperation = buildUpdateMetadataOperation(
|
||||
currentSavedFlow?.data as Record<string, unknown> | null | undefined,
|
||||
flowToSave?.data as Record<string, unknown> | null | undefined,
|
||||
);
|
||||
if (metadataOperation) {
|
||||
flowStore.onCollaborationOperations?.([metadataOperation]);
|
||||
}
|
||||
if (flowStore.flushCollaborationSave) {
|
||||
await flowStore.flushCollaborationSave();
|
||||
} else {
|
||||
syncSavedFlowStateFromCanvas();
|
||||
}
|
||||
setSaveLoading(false);
|
||||
return;
|
||||
} catch (error) {
|
||||
setSaveLoading(false);
|
||||
const detail =
|
||||
error instanceof Error
|
||||
? error.message
|
||||
: "Unknown collaboration save error";
|
||||
setErrorData({
|
||||
title: t("errors.failedToSaveFlow"),
|
||||
list: [detail],
|
||||
});
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
if (
|
||||
customStringify(flow || currentFlow) !== customStringify(currentSavedFlow)
|
||||
) {
|
||||
|
||||
@ -0,0 +1,54 @@
|
||||
import { useTranslation } from "react-i18next";
|
||||
import { Label } from "@/components/ui/label";
|
||||
import { Switch } from "@/components/ui/switch";
|
||||
import { cn } from "@/utils/utils";
|
||||
|
||||
type CollaborationBetaToggleProps = {
|
||||
enabled: boolean;
|
||||
onEnabledChange: (enabled: boolean) => void;
|
||||
disabled?: boolean;
|
||||
className?: string;
|
||||
};
|
||||
|
||||
export default function CollaborationBetaToggle({
|
||||
enabled,
|
||||
onEnabledChange,
|
||||
disabled = false,
|
||||
className,
|
||||
}: CollaborationBetaToggleProps): JSX.Element {
|
||||
const { t } = useTranslation();
|
||||
|
||||
return (
|
||||
<div
|
||||
className={cn(
|
||||
"flex max-w-xs items-start gap-2 rounded-md border bg-background px-3 py-2 shadow-sm",
|
||||
className,
|
||||
)}
|
||||
data-testid="collaboration-beta-toggle"
|
||||
>
|
||||
<Switch
|
||||
id="collaboration-operation-beta"
|
||||
checked={enabled}
|
||||
disabled={disabled}
|
||||
onCheckedChange={onEnabledChange}
|
||||
data-testid="collaboration-beta-switch"
|
||||
/>
|
||||
<div className="space-y-0.5">
|
||||
<Label
|
||||
htmlFor="collaboration-operation-beta"
|
||||
className="text-xs font-medium leading-none"
|
||||
>
|
||||
{t("flow.collaboration.betaToggleLabel", {
|
||||
defaultValue: "Beta: operation-based collaborative editing",
|
||||
})}
|
||||
</Label>
|
||||
<p className="text-[11px] text-muted-foreground">
|
||||
{t("flow.collaboration.betaToggleDescription", {
|
||||
defaultValue:
|
||||
"Uses realtime operation batches instead of full-flow autosave. Turning this on reloads the flow from the server.",
|
||||
})}
|
||||
</p>
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@ -0,0 +1,36 @@
|
||||
import { render, screen } from "@testing-library/react";
|
||||
|
||||
import CollaborationPresence from "../index";
|
||||
|
||||
describe("CollaborationPresence", () => {
|
||||
it("renders other collaborators and hides the current user", () => {
|
||||
render(
|
||||
<CollaborationPresence
|
||||
currentUserId="user-1"
|
||||
users={[
|
||||
{ user_id: "user-1", username: "ana" },
|
||||
{ user_id: "user-2", username: "bob", profile_image: null },
|
||||
]}
|
||||
/>,
|
||||
);
|
||||
|
||||
expect(screen.getByTestId("collaboration-presence")).toBeInTheDocument();
|
||||
expect(
|
||||
screen.getByTestId("collaboration-presence-user-user-2"),
|
||||
).toBeInTheDocument();
|
||||
expect(
|
||||
screen.queryByTestId("collaboration-presence-user-user-1"),
|
||||
).not.toBeInTheDocument();
|
||||
});
|
||||
|
||||
it("renders nothing when only the current user is present", () => {
|
||||
const { container } = render(
|
||||
<CollaborationPresence
|
||||
currentUserId="user-1"
|
||||
users={[{ user_id: "user-1", username: "ana" }]}
|
||||
/>,
|
||||
);
|
||||
|
||||
expect(container).toBeEmptyDOMElement();
|
||||
});
|
||||
});
|
||||
@ -0,0 +1,68 @@
|
||||
import { useTranslation } from "react-i18next";
|
||||
import { BASE_URL_API } from "@/customization/config-constants";
|
||||
import type { CollaborationPresenceUser } from "@/types/flow-collaboration";
|
||||
import { cn } from "@/utils/utils";
|
||||
|
||||
type CollaborationPresenceProps = {
|
||||
users: CollaborationPresenceUser[];
|
||||
currentUserId?: string | null;
|
||||
className?: string;
|
||||
};
|
||||
|
||||
function profileImageUrl(profileImage?: string | null): string {
|
||||
return `${BASE_URL_API}files/profile_pictures/${
|
||||
profileImage ?? "Space/046-rocket.svg"
|
||||
}`;
|
||||
}
|
||||
|
||||
export default function CollaborationPresence({
|
||||
users,
|
||||
currentUserId,
|
||||
className,
|
||||
}: CollaborationPresenceProps): JSX.Element | null {
|
||||
const { t } = useTranslation();
|
||||
|
||||
const collaborators = users.filter((user) => user.user_id !== currentUserId);
|
||||
|
||||
if (collaborators.length === 0) {
|
||||
return null;
|
||||
}
|
||||
|
||||
const visibleUsers = collaborators.slice(0, 5);
|
||||
const overflowCount = collaborators.length - visibleUsers.length;
|
||||
|
||||
return (
|
||||
<div
|
||||
className={cn("flex items-center gap-2", className)}
|
||||
data-testid="collaboration-presence"
|
||||
title={t("flow.collaboration.activeCollaborators", {
|
||||
count: collaborators.length,
|
||||
defaultValue: "{{count}} collaborators editing",
|
||||
})}
|
||||
>
|
||||
<span className="text-xs text-muted-foreground">
|
||||
{t("flow.collaboration.editing", { defaultValue: "Editing" })}
|
||||
</span>
|
||||
<div className="flex -space-x-2">
|
||||
{visibleUsers.map((user) => (
|
||||
<img
|
||||
key={user.user_id}
|
||||
src={profileImageUrl(user.profile_image)}
|
||||
alt={user.username}
|
||||
title={user.username}
|
||||
className="h-7 w-7 rounded-full border-2 border-background object-cover"
|
||||
data-testid={`collaboration-presence-user-${user.user_id}`}
|
||||
/>
|
||||
))}
|
||||
{overflowCount > 0 ? (
|
||||
<span
|
||||
className="flex h-7 w-7 items-center justify-center rounded-full border-2 border-background bg-muted text-[10px] font-medium text-muted-foreground"
|
||||
data-testid="collaboration-presence-overflow"
|
||||
>
|
||||
+{overflowCount}
|
||||
</span>
|
||||
) : null}
|
||||
</div>
|
||||
</div>
|
||||
);
|
||||
}
|
||||
@ -32,13 +32,19 @@ import { getURL } from "@/controllers/API/helpers/constants";
|
||||
import { useGetBuildsQuery } from "@/controllers/API/queries/_builds";
|
||||
import CustomLoader from "@/customization/components/custom-loader";
|
||||
import { track } from "@/customization/utils/analytics";
|
||||
import {
|
||||
buildGraphDiffOperations,
|
||||
buildUpdateNodesOperation,
|
||||
} from "@/hooks/flows/flow-operation-diff";
|
||||
import useApplyFlowToCanvas from "@/hooks/flows/use-apply-flow-to-canvas";
|
||||
import useAutoSaveFlow from "@/hooks/flows/use-autosave-flow";
|
||||
import { useFlowCollaborationEditing } from "@/hooks/flows/use-flow-collaboration-editing";
|
||||
import { useFlowEvents } from "@/hooks/flows/use-flow-events";
|
||||
import useUploadFlow from "@/hooks/flows/use-upload-flow";
|
||||
import { useAddComponent } from "@/hooks/use-add-component";
|
||||
import InspectionPanel from "@/pages/FlowPage/components/InspectionPanel";
|
||||
import useAssistantManagerStore from "@/stores/assistantManagerStore";
|
||||
import useAuthStore from "@/stores/authStore";
|
||||
import useFlowBuilderWelcomeStore from "@/stores/flowBuilderWelcomeStore";
|
||||
import { nodeColorsName } from "@/utils/styleUtils";
|
||||
import { isSupportedNodeTypes } from "@/utils/utils";
|
||||
@ -66,6 +72,8 @@ import {
|
||||
validateSelection,
|
||||
} from "../../../../utils/reactflowUtils";
|
||||
import { edgeTypes, nodeTypes } from "../../consts";
|
||||
import CollaborationBetaToggle from "../CollaborationBetaToggle";
|
||||
import CollaborationPresence from "../CollaborationPresence";
|
||||
import ConnectionLineComponent from "../ConnectionLineComponent";
|
||||
import FlowBuildingComponent from "../flowBuildingComponent";
|
||||
import SelectionMenu from "../SelectionMenuComponent";
|
||||
@ -152,6 +160,21 @@ export default function Page({
|
||||
const [lastSelection, setLastSelection] =
|
||||
useState<OnSelectionChangeParams | null>(null);
|
||||
const currentFlowId = useFlowsManagerStore((state) => state.currentFlowId);
|
||||
const userData = useAuthStore((state) => state.userData);
|
||||
const collaborationOperationMode = useFlowStore(
|
||||
(state) => state.collaborationOperationMode,
|
||||
);
|
||||
const onCollaborationOperations = useFlowStore(
|
||||
(state) => state.onCollaborationOperations,
|
||||
);
|
||||
|
||||
const {
|
||||
betaEnabled: collaborationBetaEnabled,
|
||||
setBetaEnabled: setCollaborationBetaEnabled,
|
||||
users: collaborationUsers,
|
||||
} = useFlowCollaborationEditing({
|
||||
flowId: currentFlowId || undefined,
|
||||
});
|
||||
|
||||
const { isAgentWorking, events, lastSettledAt, clearEvents } = useFlowEvents(
|
||||
currentFlowId || undefined,
|
||||
@ -468,8 +491,31 @@ export default function Page({
|
||||
track("Component Deleted", { componentType: n.data.type });
|
||||
});
|
||||
}
|
||||
deleteNode(lastSelection.nodes.map((node) => node.id));
|
||||
deleteEdge(lastSelection.edges.map((edge) => edge.id));
|
||||
if (collaborationOperationMode && onCollaborationOperations) {
|
||||
const prevNodes = useFlowStore.getState().nodes;
|
||||
const prevEdges = useFlowStore.getState().edges;
|
||||
deleteNode(
|
||||
lastSelection.nodes.map((node) => node.id),
|
||||
{ skipCollaborationEmit: true },
|
||||
);
|
||||
deleteEdge(
|
||||
lastSelection.edges.map((edge) => edge.id),
|
||||
{ skipCollaborationEmit: true },
|
||||
);
|
||||
const nextState = useFlowStore.getState();
|
||||
const operations = buildGraphDiffOperations(
|
||||
prevNodes,
|
||||
prevEdges,
|
||||
nextState.nodes,
|
||||
nextState.edges,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
onCollaborationOperations(operations);
|
||||
}
|
||||
} else {
|
||||
deleteNode(lastSelection.nodes.map((node) => node.id));
|
||||
deleteEdge(lastSelection.edges.map((edge) => edge.id));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@ -557,20 +603,28 @@ export default function Page({
|
||||
|
||||
const onNodeDragStop: OnNodeDrag = useCallback(
|
||||
(_, node) => {
|
||||
// 👇 make moving the canvas undoable
|
||||
autoSaveFlow();
|
||||
if (collaborationOperationMode && onCollaborationOperations) {
|
||||
const movedNodes = nodes.filter(
|
||||
(canvasNode) => canvasNode.selected || canvasNode.id === node.id,
|
||||
);
|
||||
if (movedNodes.length > 0) {
|
||||
onCollaborationOperations([buildUpdateNodesOperation(movedNodes)]);
|
||||
}
|
||||
} else {
|
||||
autoSaveFlow();
|
||||
}
|
||||
updateCurrentFlow({ nodes });
|
||||
setPositionDictionary({});
|
||||
setIsDragging(false);
|
||||
setHelperLines({});
|
||||
},
|
||||
[
|
||||
takeSnapshot,
|
||||
autoSaveFlow,
|
||||
collaborationOperationMode,
|
||||
nodes,
|
||||
edges,
|
||||
reactFlowInstance,
|
||||
onCollaborationOperations,
|
||||
setPositionDictionary,
|
||||
updateCurrentFlow,
|
||||
],
|
||||
);
|
||||
|
||||
@ -863,19 +917,6 @@ export default function Page({
|
||||
? (nodes.find((n) => n.id === selectedNodeId) as AllNodeType)
|
||||
: null;
|
||||
|
||||
// Determine if InspectionPanel should be visible
|
||||
const showInspectionPanel = inspectionPanelVisible && !!selectedNode;
|
||||
|
||||
// Handler to close the inspection panel by deselecting all nodes
|
||||
const handleCloseInspectionPanel = useCallback(() => {
|
||||
setNodes((nds) =>
|
||||
nds.map((node) => ({
|
||||
...node,
|
||||
selected: false,
|
||||
})),
|
||||
);
|
||||
}, [setNodes]);
|
||||
|
||||
useEffect(() => {
|
||||
if (inspectionPanelVisible) {
|
||||
setSelectionMenuVisible(false);
|
||||
@ -894,6 +935,23 @@ export default function Page({
|
||||
isAgentWorking={isAgentWorking}
|
||||
/>
|
||||
{!isPreviewActive && <FlowToolbar />}
|
||||
{!isPreviewActive && (
|
||||
<CollaborationBetaToggle
|
||||
enabled={collaborationBetaEnabled}
|
||||
disabled={effectiveLocked}
|
||||
className="absolute right-4 top-16 z-10"
|
||||
onEnabledChange={(enabled) => {
|
||||
void setCollaborationBetaEnabled(enabled);
|
||||
}}
|
||||
/>
|
||||
)}
|
||||
{!isPreviewActive && collaborationBetaEnabled ? (
|
||||
<CollaborationPresence
|
||||
users={collaborationUsers}
|
||||
currentUserId={userData?.id}
|
||||
className="absolute right-4 top-[7.5rem] z-10 rounded-md border bg-background px-2 py-1 shadow-sm"
|
||||
/>
|
||||
) : null}
|
||||
{inspectionPanelVisible && (
|
||||
<InspectionPanel selectedNode={selectedNode} />
|
||||
)}
|
||||
|
||||
@ -19,6 +19,8 @@ import {
|
||||
trackDataLoaded,
|
||||
trackFlowBuild,
|
||||
} from "@/customization/utils/analytics";
|
||||
import type { FlowMutationOptions } from "@/hooks/flows/flow-operation-adapter";
|
||||
import { buildGraphDiffOperations } from "@/hooks/flows/flow-operation-diff";
|
||||
import { brokenEdgeMessage } from "@/utils/utils";
|
||||
import { BuildStatus, EventDeliveryType } from "../constants/enums";
|
||||
import i18n from "../i18n";
|
||||
@ -31,6 +33,7 @@ import type {
|
||||
sourceHandleType,
|
||||
targetHandleType,
|
||||
} from "../types/flow";
|
||||
import type { FlowOperation } from "../types/flow-operations";
|
||||
import type {
|
||||
ComponentsToUpdateType,
|
||||
FlowStoreType,
|
||||
@ -136,6 +139,10 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
}
|
||||
},
|
||||
autoSaveFlow: undefined,
|
||||
collaborationOperationMode: false,
|
||||
isApplyingRemoteOperations: false,
|
||||
onCollaborationOperations: undefined,
|
||||
flushCollaborationSave: undefined,
|
||||
componentsToUpdate: [],
|
||||
setComponentsToUpdate: (change) => {
|
||||
const newChange =
|
||||
@ -408,7 +415,9 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
edges: applyEdgeChanges(changes, get().edges),
|
||||
});
|
||||
},
|
||||
setNodes: (change) => {
|
||||
setNodes: (change, options?: FlowMutationOptions) => {
|
||||
const prevNodes = get().nodes;
|
||||
const prevEdges = get().edges;
|
||||
const newChange =
|
||||
typeof change === "function" ? change(get().nodes) : change;
|
||||
const { edges: newEdges } = cleanEdges(newChange, get().edges);
|
||||
@ -423,11 +432,28 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
hasIO: inputs.length > 0 || outputs.length > 0,
|
||||
});
|
||||
get().updateCurrentFlow({ nodes: newChange, edges: newEdges });
|
||||
if (get().autoSaveFlow) {
|
||||
if (
|
||||
get().collaborationOperationMode &&
|
||||
!get().isApplyingRemoteOperations &&
|
||||
!options?.skipCollaborationEmit
|
||||
) {
|
||||
const operations = buildGraphDiffOperations(
|
||||
prevNodes,
|
||||
prevEdges,
|
||||
newChange,
|
||||
newEdges,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
get().onCollaborationOperations?.(operations);
|
||||
}
|
||||
}
|
||||
if (get().autoSaveFlow && !get().collaborationOperationMode) {
|
||||
get().autoSaveFlow!();
|
||||
}
|
||||
},
|
||||
setEdges: (change) => {
|
||||
setEdges: (change, options?: FlowMutationOptions) => {
|
||||
const prevNodes = get().nodes;
|
||||
const prevEdges = get().edges;
|
||||
const newChange =
|
||||
typeof change === "function" ? change(get().edges) : change;
|
||||
set({
|
||||
@ -435,7 +461,22 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
flowState: undefined,
|
||||
});
|
||||
get().updateCurrentFlow({ edges: newChange });
|
||||
if (get().autoSaveFlow) {
|
||||
if (
|
||||
get().collaborationOperationMode &&
|
||||
!get().isApplyingRemoteOperations &&
|
||||
!options?.skipCollaborationEmit
|
||||
) {
|
||||
const operations = buildGraphDiffOperations(
|
||||
prevNodes,
|
||||
prevEdges,
|
||||
get().nodes,
|
||||
newChange,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
get().onCollaborationOperations?.(operations);
|
||||
}
|
||||
}
|
||||
if (get().autoSaveFlow && !get().collaborationOperationMode) {
|
||||
get().autoSaveFlow!();
|
||||
}
|
||||
},
|
||||
@ -444,11 +485,15 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
change: AllNodeType | ((oldState: AllNodeType) => AllNodeType),
|
||||
isUserChange: boolean = true,
|
||||
callback?: () => void,
|
||||
options?: FlowMutationOptions,
|
||||
) => {
|
||||
if (!get().nodes.find((node) => node.id === id)) {
|
||||
throw new Error("Node not found");
|
||||
}
|
||||
|
||||
const prevNodes = get().nodes;
|
||||
const prevEdges = get().edges;
|
||||
|
||||
const newChange =
|
||||
typeof change === "function"
|
||||
? change(get().nodes.find((node) => node.id === id)!)
|
||||
@ -480,14 +525,29 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
};
|
||||
});
|
||||
get().updateCurrentFlow({ nodes: newNodes, edges: newEdges });
|
||||
if (get().autoSaveFlow) {
|
||||
if (
|
||||
get().collaborationOperationMode &&
|
||||
!get().isApplyingRemoteOperations &&
|
||||
!options?.skipCollaborationEmit
|
||||
) {
|
||||
const operations = buildGraphDiffOperations(
|
||||
prevNodes,
|
||||
prevEdges,
|
||||
newNodes,
|
||||
newEdges,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
get().onCollaborationOperations?.(operations);
|
||||
}
|
||||
}
|
||||
if (get().autoSaveFlow && !get().collaborationOperationMode) {
|
||||
get().autoSaveFlow!();
|
||||
}
|
||||
},
|
||||
getNode: (id: string) => {
|
||||
return get().nodes.find((node) => node.id === id);
|
||||
},
|
||||
deleteNode: (nodeId) => {
|
||||
deleteNode: (nodeId, options?: FlowMutationOptions) => {
|
||||
const { filteredNodes, deletedNode } = get().nodes.reduce<{
|
||||
filteredNodes: AllNodeType[];
|
||||
deletedNode: AllNodeType | null;
|
||||
@ -509,7 +569,7 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
{ filteredNodes: [], deletedNode: null },
|
||||
);
|
||||
|
||||
get().setNodes(filteredNodes);
|
||||
get().setNodes(filteredNodes, options);
|
||||
|
||||
// Clear rightClickedNodeId if the deleted node was right-clicked
|
||||
const rightClickedNodeId = get().rightClickedNodeId;
|
||||
@ -528,13 +588,14 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
track("Component Deleted", { componentType: deletedNode.data.type });
|
||||
}
|
||||
},
|
||||
deleteEdge: (edgeId) => {
|
||||
deleteEdge: (edgeId, options?: FlowMutationOptions) => {
|
||||
get().setEdges(
|
||||
get().edges.filter((edge) =>
|
||||
typeof edgeId === "string"
|
||||
? edge.id !== edgeId
|
||||
: !edgeId.includes(edge.id),
|
||||
),
|
||||
options,
|
||||
);
|
||||
track("Component Connection Deleted", { edgeId });
|
||||
},
|
||||
@ -577,6 +638,8 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
const idsMap = {};
|
||||
let newNodes: AllNodeType[] = get().nodes;
|
||||
let newEdges = get().edges;
|
||||
const addedNodes: AllNodeType[] = [];
|
||||
const addedEdges: EdgeType[] = [];
|
||||
selection.nodes.forEach((node: Node) => {
|
||||
if (node.position.y < minimumY) {
|
||||
minimumY = node.position.y;
|
||||
@ -634,11 +697,13 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
);
|
||||
|
||||
// Add the new node to the list of nodes in state
|
||||
const pastedNode = { ...newNode, selected: true } as AllNodeType;
|
||||
addedNodes.push(pastedNode);
|
||||
newNodes = newNodes
|
||||
.map((node) => ({ ...node, selected: false }))
|
||||
.concat({ ...newNode, selected: true });
|
||||
.concat(pastedNode);
|
||||
});
|
||||
get().setNodes(newNodes);
|
||||
get().setNodes(newNodes, { skipCollaborationEmit: true });
|
||||
|
||||
selection.edges.forEach((edge: EdgeType) => {
|
||||
const source = idsMap[edge.source];
|
||||
@ -667,42 +732,72 @@ const useFlowStore = create<FlowStoreType>((set, get) => ({
|
||||
};
|
||||
|
||||
const id = getHandleId(source, sourceHandle, target, targetHandle);
|
||||
const pastedEdge = {
|
||||
source,
|
||||
target,
|
||||
sourceHandle,
|
||||
targetHandle,
|
||||
id,
|
||||
data: cloneDeep(edge.data),
|
||||
selected: false,
|
||||
} as EdgeType;
|
||||
addedEdges.push(pastedEdge);
|
||||
newEdges = addEdge(
|
||||
{
|
||||
source,
|
||||
target,
|
||||
sourceHandle,
|
||||
targetHandle,
|
||||
id,
|
||||
data: cloneDeep(edge.data),
|
||||
selected: false,
|
||||
},
|
||||
pastedEdge,
|
||||
newEdges.map((edge) => ({ ...edge, selected: false })),
|
||||
);
|
||||
});
|
||||
get().setEdges(newEdges);
|
||||
get().setEdges(newEdges, { skipCollaborationEmit: true });
|
||||
|
||||
if (
|
||||
get().collaborationOperationMode &&
|
||||
!get().isApplyingRemoteOperations &&
|
||||
(addedNodes.length > 0 || addedEdges.length > 0)
|
||||
) {
|
||||
const operations: FlowOperation[] = [];
|
||||
if (addedNodes.length > 0) {
|
||||
operations.push({ type: "add_nodes", nodes: addedNodes });
|
||||
}
|
||||
if (addedEdges.length > 0) {
|
||||
operations.push({ type: "add_edges", edges: addedEdges });
|
||||
}
|
||||
get().onCollaborationOperations?.(operations);
|
||||
}
|
||||
},
|
||||
setLastCopiedSelection: (newSelection, isCrop = false) => {
|
||||
if (isCrop) {
|
||||
const prevNodes = get().nodes;
|
||||
const prevEdges = get().edges;
|
||||
const nodesIdsSelected = newSelection!.nodes.map((node) => node.id);
|
||||
const edgesIdsSelected = newSelection!.edges.map((edge) => edge.id);
|
||||
|
||||
nodesIdsSelected.forEach((id) => {
|
||||
get().deleteNode(id);
|
||||
});
|
||||
|
||||
edgesIdsSelected.forEach((id) => {
|
||||
get().deleteEdge(id);
|
||||
});
|
||||
|
||||
const newNodes = get().nodes.filter(
|
||||
const newNodes = prevNodes.filter(
|
||||
(node) => !nodesIdsSelected.includes(node.id),
|
||||
);
|
||||
const newEdges = get().edges.filter(
|
||||
(edge) => !edgesIdsSelected.includes(edge.id),
|
||||
const newEdges = prevEdges.filter(
|
||||
(edge) =>
|
||||
!edgesIdsSelected.includes(edge.id) &&
|
||||
!nodesIdsSelected.includes(edge.source) &&
|
||||
!nodesIdsSelected.includes(edge.target),
|
||||
);
|
||||
|
||||
set({ nodes: newNodes, edges: newEdges });
|
||||
get().setNodes(newNodes, { skipCollaborationEmit: true });
|
||||
get().setEdges(newEdges, { skipCollaborationEmit: true });
|
||||
|
||||
if (
|
||||
get().collaborationOperationMode &&
|
||||
!get().isApplyingRemoteOperations
|
||||
) {
|
||||
const operations = buildGraphDiffOperations(
|
||||
prevNodes,
|
||||
prevEdges,
|
||||
newNodes,
|
||||
newEdges,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
get().onCollaborationOperations?.(operations);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
set({ lastCopiedSelection: newSelection });
|
||||
|
||||
@ -1,7 +1,8 @@
|
||||
import { cloneDeep } from "lodash";
|
||||
import { create } from "zustand";
|
||||
import { SAVE_DEBOUNCE_TIME } from "@/constants/constants";
|
||||
import type { FlowType } from "../types/flow";
|
||||
import { buildGraphDiffOperations } from "@/hooks/flows/flow-operation-diff";
|
||||
import type { AllNodeType, EdgeType, FlowType } from "../types/flow";
|
||||
import type {
|
||||
FlowsManagerStoreType,
|
||||
UseUndoRedoOptions,
|
||||
@ -14,8 +15,14 @@ const defaultOptions: UseUndoRedoOptions = {
|
||||
enableShortcuts: true,
|
||||
};
|
||||
|
||||
const past = {};
|
||||
const future = {};
|
||||
const past: Record<
|
||||
string,
|
||||
Array<{ nodes: AllNodeType[]; edges: EdgeType[] }>
|
||||
> = {};
|
||||
const future: Record<
|
||||
string,
|
||||
Array<{ nodes: AllNodeType[]; edges: EdgeType[] }>
|
||||
> = {};
|
||||
|
||||
const useFlowsManagerStore = create<FlowsManagerStoreType>((set, get) => ({
|
||||
IOModalOpen: false,
|
||||
@ -87,6 +94,12 @@ const useFlowsManagerStore = create<FlowsManagerStoreType>((set, get) => ({
|
||||
|
||||
future[currentFlowId] = [];
|
||||
},
|
||||
clearUndoRedoHistory: (flowId?: string) => {
|
||||
const targetFlowId = flowId ?? get().currentFlowId;
|
||||
if (!targetFlowId) return;
|
||||
past[targetFlowId] = [];
|
||||
future[targetFlowId] = [];
|
||||
},
|
||||
undo: () => {
|
||||
const newState = useFlowStore.getState();
|
||||
const currentFlowId = get().currentFlowId;
|
||||
@ -102,8 +115,20 @@ const useFlowsManagerStore = create<FlowsManagerStoreType>((set, get) => ({
|
||||
edges: newState.edges,
|
||||
});
|
||||
|
||||
newState.setNodes(pastState.nodes);
|
||||
newState.setEdges(pastState.edges);
|
||||
newState.setNodes(pastState.nodes, { skipCollaborationEmit: true });
|
||||
newState.setEdges(pastState.edges, { skipCollaborationEmit: true });
|
||||
|
||||
if (newState.collaborationOperationMode) {
|
||||
const operations = buildGraphDiffOperations(
|
||||
newState.nodes,
|
||||
newState.edges,
|
||||
pastState.nodes,
|
||||
pastState.edges,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
newState.onCollaborationOperations?.(operations);
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
redo: () => {
|
||||
@ -121,8 +146,20 @@ const useFlowsManagerStore = create<FlowsManagerStoreType>((set, get) => ({
|
||||
edges: newState.edges,
|
||||
});
|
||||
|
||||
newState.setNodes(futureState.nodes);
|
||||
newState.setEdges(futureState.edges);
|
||||
newState.setNodes(futureState.nodes, { skipCollaborationEmit: true });
|
||||
newState.setEdges(futureState.edges, { skipCollaborationEmit: true });
|
||||
|
||||
if (newState.collaborationOperationMode) {
|
||||
const operations = buildGraphDiffOperations(
|
||||
newState.nodes,
|
||||
newState.edges,
|
||||
futureState.nodes,
|
||||
futureState.edges,
|
||||
);
|
||||
if (operations.length > 0) {
|
||||
newState.onCollaborationOperations?.(operations);
|
||||
}
|
||||
}
|
||||
}
|
||||
},
|
||||
searchFlowsComponents: "",
|
||||
|
||||
@ -42,6 +42,10 @@ export type FlowOperation =
|
||||
| DeleteEdgesOp
|
||||
| UpdateMetadataOp;
|
||||
|
||||
export type FlowMutationOptions = {
|
||||
skipCollaborationEmit?: boolean;
|
||||
};
|
||||
|
||||
export type FlowOperationSubmitPayload = {
|
||||
request_id: string;
|
||||
base_revision: number;
|
||||
|
||||
@ -6,7 +6,12 @@ import type {
|
||||
ReactFlowInstance,
|
||||
Viewport,
|
||||
} from "@xyflow/react";
|
||||
import type { groupedObjType } from "@/types/components";
|
||||
import type { AllNodeType, EdgeType, FlowType } from "@/types/flow";
|
||||
import type {
|
||||
FlowMutationOptions,
|
||||
FlowOperation,
|
||||
} from "@/types/flow-operations";
|
||||
import type { BuildStatus, EventDeliveryType } from "../../../constants/enums";
|
||||
import type { LogsLogType, VertexBuildTypeAPI } from "../../api";
|
||||
import type { ChatInputType, ChatOutputType } from "../../chat";
|
||||
@ -17,8 +22,8 @@ export type FlowPoolObjectType = {
|
||||
valid: boolean;
|
||||
messages: Array<ChatOutputType | ChatInputType> | [];
|
||||
data: {
|
||||
artifacts: any | ChatOutputType | ChatInputType;
|
||||
results: any | ChatOutputType | ChatInputType;
|
||||
artifacts: unknown | ChatOutputType | ChatInputType;
|
||||
results: unknown | ChatOutputType | ChatInputType;
|
||||
};
|
||||
duration?: string;
|
||||
progress?: number;
|
||||
@ -33,8 +38,8 @@ export type FlowPoolObjectTypeNew = {
|
||||
timestamp: string;
|
||||
valid: boolean;
|
||||
data: {
|
||||
outputs?: any | ChatOutputType | ChatInputType;
|
||||
results: any | ChatOutputType | ChatInputType;
|
||||
outputs?: unknown | ChatOutputType | ChatInputType;
|
||||
results: unknown | ChatOutputType | ChatInputType;
|
||||
};
|
||||
duration?: string;
|
||||
progress?: number;
|
||||
@ -76,6 +81,12 @@ export type FlowStoreType = {
|
||||
}) => void;
|
||||
fitViewNode: (nodeId: string) => void;
|
||||
autoSaveFlow: ((flow?: FlowType) => void) | undefined;
|
||||
collaborationOperationMode: boolean;
|
||||
isApplyingRemoteOperations: boolean;
|
||||
onCollaborationOperations:
|
||||
| ((operations: FlowOperation[]) => void)
|
||||
| undefined;
|
||||
flushCollaborationSave: (() => Promise<void>) | undefined;
|
||||
componentsToUpdate: ComponentsToUpdateType[];
|
||||
setComponentsToUpdate: (
|
||||
update:
|
||||
@ -142,31 +153,40 @@ export type FlowStoreType = {
|
||||
onEdgesChange: OnEdgesChange<EdgeType>;
|
||||
setNodes: (
|
||||
update: AllNodeType[] | ((oldState: AllNodeType[]) => AllNodeType[]),
|
||||
options?: FlowMutationOptions,
|
||||
) => void;
|
||||
setEdges: (
|
||||
update: EdgeType[] | ((oldState: EdgeType[]) => EdgeType[]),
|
||||
options?: FlowMutationOptions,
|
||||
) => void;
|
||||
setNode: (
|
||||
id: string,
|
||||
update: AllNodeType | ((oldState: AllNodeType) => AllNodeType),
|
||||
isUserChange?: boolean,
|
||||
callback?: () => void,
|
||||
options?: FlowMutationOptions,
|
||||
) => void;
|
||||
getNode: (id: string) => AllNodeType | undefined;
|
||||
deleteNode: (nodeId: string | Array<string>) => void;
|
||||
deleteEdge: (edgeId: string | Array<string>) => void;
|
||||
deleteNode: (
|
||||
nodeId: string | Array<string>,
|
||||
options?: FlowMutationOptions,
|
||||
) => void;
|
||||
deleteEdge: (
|
||||
edgeId: string | Array<string>,
|
||||
options?: FlowMutationOptions,
|
||||
) => void;
|
||||
paste: (
|
||||
selection: { nodes: any; edges: any },
|
||||
selection: { nodes: AllNodeType[]; edges: EdgeType[] },
|
||||
position: { x: number; y: number; paneX?: number; paneY?: number },
|
||||
) => void;
|
||||
lastCopiedSelection: { nodes: any; edges: any } | null;
|
||||
lastCopiedSelection: { nodes: AllNodeType[]; edges: EdgeType[] } | null;
|
||||
setLastCopiedSelection: (
|
||||
newSelection: { nodes: any; edges: any } | null,
|
||||
newSelection: { nodes: AllNodeType[]; edges: EdgeType[] } | null,
|
||||
isCrop?: boolean,
|
||||
) => void;
|
||||
cleanFlow: () => void;
|
||||
setFilterEdge: (newState) => void;
|
||||
getFilterEdge: any[];
|
||||
setFilterEdge: (newState: groupedObjType[]) => void;
|
||||
getFilterEdge: groupedObjType[];
|
||||
setFilterComponent: (newState) => void;
|
||||
getFilterComponent: string;
|
||||
rightClickedNodeId: string | null;
|
||||
|
||||
@ -15,6 +15,7 @@ export type FlowsManagerStoreType = {
|
||||
undo: () => void;
|
||||
redo: () => void;
|
||||
takeSnapshot: () => void;
|
||||
clearUndoRedoHistory: (flowId?: string) => void;
|
||||
examples: Array<FlowType>;
|
||||
setExamples: (examples: FlowType[]) => void;
|
||||
setCurrentFlow: (flow?: FlowType) => void;
|
||||
|
||||
Reference in New Issue
Block a user