Release Flow Tools 0.2.0

2026-09-02 08:10:52 +02:00
parent 12d0ac0155
commit 57dd710275
28 changed files with 810 additions and 66 deletions
@@ -1,4 +1,4 @@
-import { useMemo, useState } from "react";
+import { useMemo, useRef, useState } from "react";
 import {
  stableStringify,
  stringifyCsv,
@@ -6,10 +6,12 @@ import {
  type JsonValue,
 } from "@add-ideas/toolbox-helpers";
 import { parseData, type DataFormat } from "../core/data";
+import { runFlowJob } from "../core/flow-client";
 import {
  exportRecipe,
  importRecipe,
  records,
+  renderPipelineSvg,
  runPipeline,
  type PipelineResult,
  type Stage,
@@ -48,7 +50,12 @@ export function Workbench() {
    ),
    [selected, setSelected] = useState(2),
    [status, setStatus] = useState("Sample pipeline completed locally."),
-    [recipeText, setRecipeText] = useState("");
+    [recipeText, setRecipeText] = useState(""),
+    [busy, setBusy] = useState(false),
+    [progress, setProgress] = useState(0);
+  const operation = useRef<{ revision: number; controller?: AbortController }>({
+    revision: 0,
+  });
  const snapshot =
    result.snapshots[Math.min(selected, result.snapshots.length - 1)]!;
  function parsePrimary() {
@@ -71,16 +78,42 @@ export function Workbench() {
      setStatus(`${msg(e)} Last valid secondary input retained.`);
    }
  }
-  function run() {
+  async function run() {
+    operation.current.controller?.abort();
+    const controller = new AbortController(),
+      revision = ++operation.current.revision;
+    operation.current.controller = controller;
+    setBusy(true);
+    setProgress(0);
    try {
-      const next = runPipeline(input, stages, right);
+      const next = await runFlowJob(input, stages, right, {
+        signal: controller.signal,
+        onProgress: (item) => {
+          if (operation.current.revision !== revision) return;
+          setProgress(item.total ? item.completed / item.total : 0);
+          setStatus(`Running ${item.stage} locally…`);
+        },
+      });
+      if (operation.current.revision !== revision) return;
      setResult(next);
      setSelected(next.snapshots.length - 1);
      setStatus(
        `Completed ${stages.filter((s) => s.enabled).length} stages; ${next.rows.length} rows remain.`,
      );
    } catch (e) {
+      if (operation.current.revision !== revision) return;
+      if (e instanceof DOMException && e.name === "AbortError") {
+        setStatus(
+          "Pipeline run cancelled. Last successful snapshots remain visible.",
+        );
+        return;
+      }
      setStatus(`${msg(e)} Last successful stage snapshots remain visible.`);
+    } finally {
+      if (operation.current.revision === revision) {
+        setBusy(false);
+        operation.current.controller = undefined;
+      }
    }
  }
  function update(index: number, stage: Stage) {
@@ -90,19 +123,24 @@ export function Workbench() {
    setStages((s) => [
      ...s,
      {
-        id: `stage-${Date.now().toString(36)}`,
+        id: nextStageId(s),
        type,
        enabled: true,
        config: defaults(type),
      },
    ]);
  }
-  function download(kind: "json" | "csv" | "recipe") {
+  function download(kind: "json" | "csv" | "recipe" | "svg") {
    if (kind === "recipe")
      triggerBlobDownload(
        new Blob([exportRecipe(stages)], { type: "application/json" }),
        "flow-recipe.json",
      );
+    else if (kind === "svg")
+      triggerBlobDownload(
+        new Blob([renderPipelineSvg(stages)], { type: "image/svg+xml" }),
+        "flow-pipeline.svg",
+      );
    else if (kind === "json")
      triggerBlobDownload(
        new Blob([stableStringify(result.rows, 2)], {
@@ -124,6 +162,41 @@ export function Workbench() {
      );
    }
  }
+  async function openData(file: File | undefined, secondary = false) {
+    if (!file) return;
+    try {
+      if (file.size > 2 * 1024 * 1024)
+        throw new Error("Input file exceeds the 2 MiB limit.");
+      const text = await file.text(),
+        detected = inferFormat(file.name),
+        next = records(parseData(text, detected).rows);
+      if (secondary) {
+        setRightSource(text);
+        setRight(next);
+        setStatus(`Opened ${next.length} secondary rows from ${file.name}.`);
+      } else {
+        setFormat(detected);
+        setSource(text);
+        setInput(next);
+        setStatus(`Opened ${next.length} primary rows from ${file.name}.`);
+      }
+    } catch (e) {
+      setStatus(`${msg(e)} Last valid input retained.`);
+    }
+  }
+  async function openRecipe(file: File | undefined) {
+    if (!file) return;
+    try {
+      if (file.size > 256 * 1024) throw new Error("Recipe exceeds 256 KiB.");
+      const text = await file.text(),
+        next = importRecipe(text);
+      setRecipeText(text);
+      setStages(next);
+      setStatus(`Imported ${next.length} validated stages from ${file.name}.`);
+    } catch (e) {
+      setStatus(msg(e));
+    }
+  }
  function loadRecipe() {
    try {
      const next = importRecipe(recipeText);
@@ -144,11 +217,25 @@ export function Workbench() {
            explicit losses with a deterministic recipe.
          </p>
        </div>
-        <button className="primary" onClick={run}>
-          Run complete pipeline
-        </button>
+        <div className="run-actions">
+          <button
+            className="primary"
+            onClick={() => void run()}
+            disabled={busy}
+          >
+            Run complete pipeline
+          </button>
+          {busy && (
+            <button onClick={() => operation.current.controller?.abort()}>
+              Cancel
+            </button>
+          )}
+        </div>
      </section>
      <p role="status">{status}</p>
+      {busy && (
+        <progress value={progress} max={1} aria-label="Pipeline progress" />
+      )}
      <section className="sources">
        <section className="panel">
          <h2>Primary source</h2>
@@ -169,7 +256,17 @@ export function Workbench() {
            onChange={(e) => setSource(e.target.value)}
            aria-label="Primary source"
          />
-          <button onClick={parsePrimary}>Parse primary</button>
+          <div className="exports">
+            <button onClick={parsePrimary}>Parse primary</button>
+            <label className="file-button">
+              Open data file
+              <input
+                type="file"
+                accept=".json,.csv,.ndjson,.jsonl,.xml"
+                onChange={(event) => void openData(event.target.files?.[0])}
+              />
+            </label>
+          </div>
        </section>
        <section className="panel">
          <h2>Secondary join source</h2>
@@ -179,7 +276,19 @@ export function Workbench() {
            onChange={(e) => setRightSource(e.target.value)}
            aria-label="Secondary source"
          />
-          <button onClick={parseRight}>Parse secondary</button>
+          <div className="exports">
+            <button onClick={parseRight}>Parse secondary</button>
+            <label className="file-button">
+              Open join file
+              <input
+                type="file"
+                accept=".json,.csv,.ndjson,.jsonl,.xml"
+                onChange={(event) =>
+                  void openData(event.target.files?.[0], true)
+                }
+              />
+            </label>
+          </div>
        </section>
      </section>
      <section className="pipeline">
@@ -196,6 +305,9 @@ export function Workbench() {
                <option value="group">Group</option>
                <option value="join">Join</option>
                <option value="format">Format type</option>
+                <option value="deduplicate">Remove duplicates</option>
+                <option value="slice">Slice rows</option>
+                <option value="explode">Explode array</option>
              </select>
            </label>
            <button
@@ -250,10 +362,31 @@ export function Workbench() {
            </ul>
          )}
          <Table rows={snapshot.rows} />
+          <dl className="analysis-grid" aria-label="Result analysis">
+            <div>
+              <dt>Rows</dt>
+              <dd>
+                {result.analysis.inputRows} → {result.analysis.outputRows}
+              </dd>
+            </div>
+            <div>
+              <dt>Fields</dt>
+              <dd>{result.analysis.fields}</dd>
+            </div>
+            <div>
+              <dt>Empty cells</dt>
+              <dd>{result.analysis.nullCells}</dd>
+            </div>
+            <div>
+              <dt>Input duplicates</dt>
+              <dd>{result.analysis.duplicateRows}</dd>
+            </div>
+          </dl>
          <div className="exports">
            <button onClick={() => download("json")}>Result JSON</button>
            <button onClick={() => download("csv")}>Result CSV</button>
            <button onClick={() => download("recipe")}>Recipe JSON</button>
+            <button onClick={() => download("svg")}>Pipeline SVG</button>
          </div>
          <details>
            <summary>Import deterministic recipe</summary>
@@ -263,6 +396,14 @@ export function Workbench() {
              aria-label="Recipe JSON"
            />
            <button onClick={loadRecipe}>Validate and import</button>
+            <label className="file-button">
+              Open recipe file
+              <input
+                type="file"
+                accept=".json,application/json"
+                onChange={(event) => void openRecipe(event.target.files?.[0])}
+              />
+            </label>
          </details>
        </section>
      </section>
@@ -303,8 +444,20 @@ function StageEditor({
      </header>
      <Config stage={stage} set={set} />
      <footer>
-        <button onClick={() => move(-1)}>↑</button>
-        <button onClick={() => move(1)}>↓</button>
+        <button
+          type="button"
+          aria-label="Move stage up"
+          onClick={() => move(-1)}
+        >
+          ↑
+        </button>
+        <button
+          type="button"
+          aria-label="Move stage down"
+          onClick={() => move(1)}
+        >
+          ↓
+        </button>
        <button onClick={remove}>Remove</button>
      </footer>
    </article>
@@ -415,6 +568,36 @@ function Config({
        />
      </>
    );
+  if (stage.type === "deduplicate")
+    return (
+      <Field
+        label="Identity fields (blank = whole row)"
+        value={f.fields}
+        set={(v) => set("fields", v)}
+      />
+    );
+  if (stage.type === "slice")
+    return (
+      <>
+        <Field label="Offset" value={f.offset} set={(v) => set("offset", v)} />
+        <Field label="Count" value={f.count} set={(v) => set("count", v)} />
+      </>
+    );
+  if (stage.type === "explode")
+    return (
+      <>
+        <Field
+          label="Array field"
+          value={f.field}
+          set={(v) => set("field", v)}
+        />
+        <Field
+          label="Output field"
+          value={f.target}
+          set={(v) => set("target", v)}
+        />
+      </>
+    );
  return (
    <>
      <Field label="Target" value={f.target} set={(v) => set("target", v)} />
@@ -530,5 +713,26 @@ function defaults(t: StageType): Record<string, string> {
                    mode: "left",
                    prefix: "right_",
                  }
-                : { field: "score", as: "number" };
+                : t === "format"
+                  ? { field: "score", as: "number" }
+                  : t === "deduplicate"
+                    ? { fields: "" }
+                    : t === "slice"
+                      ? { offset: "0", count: "100" }
+                      : { field: "items", target: "item" };
+}
+
+function inferFormat(fileName: string): DataFormat {
+  const extension = fileName.split(".").at(-1)?.toLowerCase();
+  if (extension === "csv") return "csv";
+  if (extension === "ndjson" || extension === "jsonl") return "ndjson";
+  if (extension === "xml") return "xml";
+  return "json";
+}
+
+function nextStageId(stages: readonly Stage[]): string {
+  const ids = new Set(stages.map((stage) => stage.id));
+  let suffix = 1;
+  while (ids.has(`stage-${suffix}`)) suffix += 1;
+  return `stage-${suffix}`;
 }