diff --git a/src/store/utils/__tests__/executionUtils.test.ts b/src/store/utils/__tests__/executionUtils.test.ts new file mode 100644 index 00000000..b296aa13 --- /dev/null +++ b/src/store/utils/__tests__/executionUtils.test.ts @@ -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; + 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).imageRef).toBe("ref"); + }); +}); + +describe("concurrency settings", () => { + let mockStorage: Record; + + 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"); + }); +}); diff --git a/src/store/utils/executionUtils.ts b/src/store/utils/executionUtils.ts new file mode 100644 index 00000000..5b66c2ca --- /dev/null +++ b/src/store/utils/executionUtils.ts @@ -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(); + const adjList = new Map(); + + 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(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; + + // 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; + }); +} diff --git a/src/store/workflowStore.ts b/src/store/workflowStore.ts index fef60625..a63ba0ab 100644 --- a/src/store/workflowStore.ts +++ b/src/store/workflowStore.ts @@ -59,6 +59,17 @@ import { GROUP_COLOR_ORDER, } from "./utils/nodeDefaults"; 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"; @@ -323,121 +334,6 @@ async function waitForPendingImageSyncs(timeout: number = 60000): Promise } } -// 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(); - const adjList = new Map(); - - 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(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; - - // 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 export { generateWorkflowId, saveGenerateImageDefaults, saveNanoBananaDefaults } from "./utils/localStorage";