weitao
/
popiart-node-canvas
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

refactor: extract execution utilities from workflow store 

Move groupNodesByLevel, chunk, revokeBlobUrl, clearNodeImageRefs,
and concurrency settings into executionUtils.ts with 21 new tests.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
handoff-20260429-1057
shrimbly 5 months ago
parent
4c3847bfcb
commit
194240133c
3 changed files with 349 additions and 115 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 203
      src/store/utils/__tests__/executionUtils.test.ts
  2. 135
      src/store/utils/executionUtils.ts
  3. 126
      src/store/workflowStore.ts

203
src/store/utils/__tests__/executionUtils.test.ts
View File

@ -0,0 +1,203 @@
import { describe, it, expect, vi, beforeEach } from "vitest";
import {
groupNodesByLevel,
chunk,
revokeBlobUrl,
clearNodeImageRefs,
loadConcurrencySetting,
saveConcurrencySetting,
DEFAULT_MAX_CONCURRENT_CALLS,
CONCURRENCY_SETTINGS_KEY,
} from "../executionUtils";
import type { WorkflowNode, WorkflowEdge } from "@/types";
function makeNode(id: string, type = "prompt"): WorkflowNode {
return {
id,
type,
position: { x: 0, y: 0 },
data: {},
} as WorkflowNode;
}
function makeEdge(source: string, target: string): WorkflowEdge {
return {
id: `${source}-${target}`,
source,
target,
sourceHandle: "text",
targetHandle: "text",
} as WorkflowEdge;
}
describe("groupNodesByLevel", () => {
it("should put all nodes at level 0 when there are no edges", () => {
const nodes = [makeNode("a"), makeNode("b"), makeNode("c")];
const result = groupNodesByLevel(nodes, []);
expect(result).toHaveLength(1);
expect(result[0].level).toBe(0);
expect(result[0].nodeIds.sort()).toEqual(["a", "b", "c"]);
});
it("should handle a linear chain", () => {
const nodes = [makeNode("a"), makeNode("b"), makeNode("c")];
const edges = [makeEdge("a", "b"), makeEdge("b", "c")];
const result = groupNodesByLevel(nodes, edges);
expect(result).toHaveLength(3);
expect(result[0].nodeIds).toEqual(["a"]);
expect(result[1].nodeIds).toEqual(["b"]);
expect(result[2].nodeIds).toEqual(["c"]);
});
it("should group parallel nodes at the same level", () => {
// a -> b, a -> c (b and c are parallel)
const nodes = [makeNode("a"), makeNode("b"), makeNode("c")];
const edges = [makeEdge("a", "b"), makeEdge("a", "c")];
const result = groupNodesByLevel(nodes, edges);
expect(result).toHaveLength(2);
expect(result[0].nodeIds).toEqual(["a"]);
expect(result[1].nodeIds.sort()).toEqual(["b", "c"]);
});
it("should handle diamond dependencies", () => {
// a -> b, a -> c, b -> d, c -> d
const nodes = [makeNode("a"), makeNode("b"), makeNode("c"), makeNode("d")];
const edges = [
makeEdge("a", "b"),
makeEdge("a", "c"),
makeEdge("b", "d"),
makeEdge("c", "d"),
];
const result = groupNodesByLevel(nodes, edges);
expect(result).toHaveLength(3);
expect(result[0].nodeIds).toEqual(["a"]);
expect(result[1].nodeIds.sort()).toEqual(["b", "c"]);
expect(result[2].nodeIds).toEqual(["d"]);
});
it("should handle empty inputs", () => {
const result = groupNodesByLevel([], []);
expect(result).toEqual([]);
});
});
describe("chunk", () => {
it("should split array into chunks of specified size", () => {
expect(chunk([1, 2, 3, 4, 5], 2)).toEqual([[1, 2], [3, 4], [5]]);
});
it("should handle array smaller than chunk size", () => {
expect(chunk([1, 2], 5)).toEqual([[1, 2]]);
});
it("should handle empty array", () => {
expect(chunk([], 3)).toEqual([]);
});
it("should handle chunk size of 1", () => {
expect(chunk([1, 2, 3], 1)).toEqual([[1], [2], [3]]);
});
it("should handle exact multiples", () => {
expect(chunk([1, 2, 3, 4], 2)).toEqual([[1, 2], [3, 4]]);
});
});
describe("revokeBlobUrl", () => {
it("should revoke blob URLs", () => {
const spy = vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => {});
revokeBlobUrl("blob:http://localhost/abc");
expect(spy).toHaveBeenCalledWith("blob:http://localhost/abc");
spy.mockRestore();
});
it("should not revoke non-blob URLs", () => {
const spy = vi.spyOn(URL, "revokeObjectURL").mockImplementation(() => {});
revokeBlobUrl("http://example.com/image.png");
expect(spy).not.toHaveBeenCalled();
spy.mockRestore();
});
it("should handle null", () => {
expect(() => revokeBlobUrl(null)).not.toThrow();
});
it("should handle undefined", () => {
expect(() => revokeBlobUrl(undefined)).not.toThrow();
});
});
describe("clearNodeImageRefs", () => {
it("should clear imageRef fields", () => {
const nodes = [
{
...makeNode("a"),
data: {
imageRef: "some-ref",
sourceImageRef: "src-ref",
outputImageRef: "out-ref",
inputImageRefs: ["ref1"],
prompt: "keep this",
},
},
] as unknown as WorkflowNode[];
const result = clearNodeImageRefs(nodes);
const data = result[0].data as Record<string, unknown>;
expect(data.imageRef).toBeUndefined();
expect(data.sourceImageRef).toBeUndefined();
expect(data.outputImageRef).toBeUndefined();
expect(data.inputImageRefs).toBeUndefined();
expect(data.prompt).toBe("keep this");
});
it("should not mutate original nodes", () => {
const original = [
{
...makeNode("a"),
data: { imageRef: "ref" },
},
] as unknown as WorkflowNode[];
clearNodeImageRefs(original);
expect((original[0].data as Record<string, unknown>).imageRef).toBe("ref");
});
});
describe("concurrency settings", () => {
let mockStorage: Record<string, string>;
beforeEach(() => {
mockStorage = {};
vi.stubGlobal("localStorage", {
getItem: (key: string) => mockStorage[key] ?? null,
setItem: (key: string, value: string) => { mockStorage[key] = value; },
});
});
it("should return default when no setting stored", () => {
expect(loadConcurrencySetting()).toBe(DEFAULT_MAX_CONCURRENT_CALLS);
});
it("should load stored setting", () => {
mockStorage[CONCURRENCY_SETTINGS_KEY] = "5";
expect(loadConcurrencySetting()).toBe(5);
});
it("should reject out-of-range values", () => {
mockStorage[CONCURRENCY_SETTINGS_KEY] = "0";
expect(loadConcurrencySetting()).toBe(DEFAULT_MAX_CONCURRENT_CALLS);
mockStorage[CONCURRENCY_SETTINGS_KEY] = "11";
expect(loadConcurrencySetting()).toBe(DEFAULT_MAX_CONCURRENT_CALLS);
});
it("should reject invalid values", () => {
mockStorage[CONCURRENCY_SETTINGS_KEY] = "abc";
expect(loadConcurrencySetting()).toBe(DEFAULT_MAX_CONCURRENT_CALLS);
});
it("should save setting", () => {
saveConcurrencySetting(7);
expect(mockStorage[CONCURRENCY_SETTINGS_KEY]).toBe("7");
});
});

135
src/store/utils/executionUtils.ts
View File

@ -0,0 +1,135 @@
/**
* Execution Utilities
*
* Pure utility functions used by the workflow execution engine.
* Extracted from workflowStore.ts for testability and reuse.
*/
import { WorkflowNode, WorkflowEdge, WorkflowNodeData } from "@/types";
// Concurrency settings
export const CONCURRENCY_SETTINGS_KEY = "node-banana-concurrency-limit";
export const DEFAULT_MAX_CONCURRENT_CALLS = 3;
/**
* Load concurrency setting from localStorage
*/
export const loadConcurrencySetting = (): number => {
if (typeof window === "undefined") return DEFAULT_MAX_CONCURRENT_CALLS;
const stored = localStorage.getItem(CONCURRENCY_SETTINGS_KEY);
if (stored) {
const parsed = parseInt(stored, 10);
if (!isNaN(parsed) && parsed >= 1 && parsed <= 10) {
return parsed;
}
}
return DEFAULT_MAX_CONCURRENT_CALLS;
};
/**
* Save concurrency setting to localStorage
*/
export const saveConcurrencySetting = (value: number): void => {
if (typeof window === "undefined") return;
localStorage.setItem(CONCURRENCY_SETTINGS_KEY, String(value));
};
/**
* Level grouping for parallel execution
*/
export interface LevelGroup {
level: number;
nodeIds: string[];
}
/**
* Groups nodes by dependency level using Kahn's algorithm variant.
* Nodes at the same level can be executed in parallel.
* Level 0 = nodes with no incoming edges (roots)
* Level N = nodes whose dependencies are all at levels < N
*/
export function groupNodesByLevel(
nodes: WorkflowNode[],
edges: WorkflowEdge[]
): LevelGroup[] {
// Calculate in-degree for each node
const inDegree = new Map<string, number>();
const adjList = new Map<string, string[]>();
nodes.forEach((n) => {
inDegree.set(n.id, 0);
adjList.set(n.id, []);
});
edges.forEach((e) => {
inDegree.set(e.target, (inDegree.get(e.target) || 0) + 1);
adjList.get(e.source)?.push(e.target);
});
// BFS with level tracking (Kahn's algorithm variant)
const levels: LevelGroup[] = [];
let currentLevel = nodes
.filter((n) => inDegree.get(n.id) === 0)
.map((n) => n.id);
let levelNum = 0;
while (currentLevel.length > 0) {
levels.push({ level: levelNum, nodeIds: [...currentLevel] });
const nextLevel: string[] = [];
for (const nodeId of currentLevel) {
for (const child of adjList.get(nodeId) || []) {
const newDegree = (inDegree.get(child) || 1) - 1;
inDegree.set(child, newDegree);
if (newDegree === 0) {
nextLevel.push(child);
}
}
}
currentLevel = nextLevel;
levelNum++;
}
return levels;
}
/**
* Chunk an array into smaller arrays of specified size
*/
export function chunk<T>(array: T[], size: number): T[][] {
const chunks: T[][] = [];
for (let i = 0; i < array.length; i += size) {
chunks.push(array.slice(i, i + size));
}
return chunks;
}
/**
* Revoke a blob URL if the value is one, to free the underlying memory.
*/
export function revokeBlobUrl(url: string | null | undefined): void {
if (url && url.startsWith('blob:')) {
try { URL.revokeObjectURL(url); } catch { /* ignore */ }
}
}
/**
* Clear all imageRefs from nodes (used when saving to a different directory)
*/
export function clearNodeImageRefs(nodes: WorkflowNode[]): WorkflowNode[] {
return nodes.map(node => {
const data = { ...node.data } as Record<string, unknown>;
// Revoke blob URLs for video outputs before clearing
revokeBlobUrl(data.outputVideo as string | undefined);
// Clear all ref fields regardless of node type
delete data.imageRef;
delete data.sourceImageRef;
delete data.outputImageRef;
delete data.inputImageRefs;
return { ...node, data: data as WorkflowNodeData } as WorkflowNode;
});
}

126
src/store/workflowStore.ts
View File

@ -59,6 +59,17 @@ import {
GROUP_COLOR_ORDER, GROUP_COLOR_ORDER,
} from "./utils/nodeDefaults"; } from "./utils/nodeDefaults";
import { buildGenerateHeaders, buildLlmHeaders } from "./utils/buildApiHeaders"; import { buildGenerateHeaders, buildLlmHeaders } from "./utils/buildApiHeaders";
import {
CONCURRENCY_SETTINGS_KEY,
loadConcurrencySetting,
saveConcurrencySetting,
groupNodesByLevel,
chunk,
revokeBlobUrl,
clearNodeImageRefs,
} from "./utils/executionUtils";
export type { LevelGroup } from "./utils/executionUtils";
export { CONCURRENCY_SETTINGS_KEY } from "./utils/executionUtils";
export type EdgeStyle = "angular" | "curved"; export type EdgeStyle = "angular" | "curved";
@ -323,121 +334,6 @@ async function waitForPendingImageSyncs(timeout: number = 60000): Promise<void>
} }
} }
// Concurrency settings
export const CONCURRENCY_SETTINGS_KEY = "node-banana-concurrency-limit";
const DEFAULT_MAX_CONCURRENT_CALLS = 3;
// Load/save concurrency setting from localStorage
const loadConcurrencySetting = (): number => {
if (typeof window === "undefined") return DEFAULT_MAX_CONCURRENT_CALLS;
const stored = localStorage.getItem(CONCURRENCY_SETTINGS_KEY);
if (stored) {
const parsed = parseInt(stored, 10);
if (!isNaN(parsed) && parsed >= 1 && parsed <= 10) {
return parsed;
}
}
return DEFAULT_MAX_CONCURRENT_CALLS;
};
const saveConcurrencySetting = (value: number): void => {
if (typeof window === "undefined") return;
localStorage.setItem(CONCURRENCY_SETTINGS_KEY, String(value));
};
// Level grouping for parallel execution
export interface LevelGroup {
level: number;
nodeIds: string[];
}
/**
* Groups nodes by dependency level using Kahn's algorithm variant.
* Nodes at the same level can be executed in parallel.
* Level 0 = nodes with no incoming edges (roots)
* Level N = nodes whose dependencies are all at levels < N
*/
function groupNodesByLevel(
nodes: WorkflowNode[],
edges: WorkflowEdge[]
): LevelGroup[] {
// Calculate in-degree for each node
const inDegree = new Map<string, number>();
const adjList = new Map<string, string[]>();
nodes.forEach((n) => {
inDegree.set(n.id, 0);
adjList.set(n.id, []);
});
edges.forEach((e) => {
inDegree.set(e.target, (inDegree.get(e.target) || 0) + 1);
adjList.get(e.source)?.push(e.target);
});
// BFS with level tracking (Kahn's algorithm variant)
const levels: LevelGroup[] = [];
let currentLevel = nodes
.filter((n) => inDegree.get(n.id) === 0)
.map((n) => n.id);
let levelNum = 0;
while (currentLevel.length > 0) {
levels.push({ level: levelNum, nodeIds: [...currentLevel] });
const nextLevel: string[] = [];
for (const nodeId of currentLevel) {
for (const child of adjList.get(nodeId) || []) {
const newDegree = (inDegree.get(child) || 1) - 1;
inDegree.set(child, newDegree);
if (newDegree === 0) {
nextLevel.push(child);
}
}
}
currentLevel = nextLevel;
levelNum++;
}
return levels;
}
/**
* Chunk an array into smaller arrays of specified size
*/
function chunk<T>(array: T[], size: number): T[][] {
const chunks: T[][] = [];
for (let i = 0; i < array.length; i += size) {
chunks.push(array.slice(i, i + size));
}
return chunks;
}
// Clear all imageRefs from nodes (used when saving to a different directory)
/** Revoke a blob URL if the value is one, to free the underlying memory. */
function revokeBlobUrl(url: string | null | undefined): void {
if (url && url.startsWith('blob:')) {
try { URL.revokeObjectURL(url); } catch { /* ignore */ }
}
}
function clearNodeImageRefs(nodes: WorkflowNode[]): WorkflowNode[] {
return nodes.map(node => {
const data = { ...node.data } as Record<string, unknown>;
// Revoke blob URLs for video outputs before clearing
revokeBlobUrl(data.outputVideo as string | undefined);
// Clear all ref fields regardless of node type
delete data.imageRef;
delete data.sourceImageRef;
delete data.outputImageRef;
delete data.inputImageRefs;
return { ...node, data: data as WorkflowNodeData } as WorkflowNode;
});
}
// Re-export for backward compatibility // Re-export for backward compatibility
export { generateWorkflowId, saveGenerateImageDefaults, saveNanoBananaDefaults } from "./utils/localStorage"; export { generateWorkflowId, saveGenerateImageDefaults, saveNanoBananaDefaults } from "./utils/localStorage";

Write Preview
Loading…
Cancel
Save