From f2839ebc6808a0445d64a41bedd1079b26ab6918 Mon Sep 17 00:00:00 2001 From: Shine1i Date: Fri, 20 Feb 2026 14:41:38 +0100 Subject: [PATCH] refactor: extract and consolidate execution runtime and tracking logic --- .../recipe-studio/executions/hydration.ts | 22 + .../recipe-studio/executions/runtime.ts | 128 +++++ .../recipe-studio/executions/tracker.ts | 320 +++++++++++ .../hooks/use-recipe-executions.ts | 515 +++--------------- 4 files changed, 534 insertions(+), 451 deletions(-) create mode 100644 studio/frontend/src/features/recipe-studio/executions/hydration.ts create mode 100644 studio/frontend/src/features/recipe-studio/executions/runtime.ts create mode 100644 studio/frontend/src/features/recipe-studio/executions/tracker.ts diff --git a/studio/frontend/src/features/recipe-studio/executions/hydration.ts b/studio/frontend/src/features/recipe-studio/executions/hydration.ts new file mode 100644 index 0000000000..08490aec21 --- /dev/null +++ b/studio/frontend/src/features/recipe-studio/executions/hydration.ts @@ -0,0 +1,22 @@ +import { listRecipeExecutions } from "../data/executions-db"; +import type { RecipeExecutionRecord } from "../execution-types"; +import { + isExecutionInProgress, + sortExecutions, + withExecutionDefaults, +} from "./execution-helpers"; + +export async function loadSortedRecipeExecutions( + recipeId: string, +): Promise { + const records = await listRecipeExecutions(recipeId); + return sortExecutions(records.map(withExecutionDefaults)); +} + +export function findResumableExecution( + records: RecipeExecutionRecord[], +): RecipeExecutionRecord | null { + return ( + records.find((record) => record.jobId && isExecutionInProgress(record.status)) ?? null + ); +} diff --git a/studio/frontend/src/features/recipe-studio/executions/runtime.ts b/studio/frontend/src/features/recipe-studio/executions/runtime.ts new file mode 100644 index 0000000000..4c4aa8e871 --- /dev/null +++ b/studio/frontend/src/features/recipe-studio/executions/runtime.ts @@ -0,0 +1,128 @@ +import type { JobEvent, JobStatusResponse } from "../api"; +import type { RecipeExecutionKind, RecipeExecutionRecord } from "../execution-types"; +import { + DATASET_PAGE_SIZE, + mapJobStatus, + normalizeObject, +} from "./execution-helpers"; + +const MAX_LOG_LINES = 1500; + +function formatEventTime(ts: unknown): string { + if (typeof ts !== "number" || !Number.isFinite(ts)) { + return new Date().toLocaleTimeString(); + } + const ms = ts > 10_000_000_000 ? ts : ts * 1000; + return new Date(ms).toLocaleTimeString(); +} + +export function appendExecutionLogLine(lines: string[], nextLine: string): string[] { + const next = [...lines, nextLine]; + if (next.length <= MAX_LOG_LINES) { + return next; + } + return next.slice(next.length - MAX_LOG_LINES); +} + +export function toExecutionLogLine(event: JobEvent): string | null { + const eventType = + typeof event.payload.type === "string" ? event.payload.type : event.event; + const ts = formatEventTime(event.payload.ts); + + if (eventType === "log") { + const message = + typeof event.payload.message === "string" ? event.payload.message.trim() : ""; + if (!message) { + return null; + } + const level = + typeof event.payload.level === "string" && event.payload.level.length > 0 + ? event.payload.level.toUpperCase() + : "INFO"; + return `[${ts}] [${level}] ${message}`; + } + + if (eventType === "job.started") { + return `[${ts}] [INFO] Job started`; + } + if (eventType === "job.completed") { + return `[${ts}] [INFO] Job completed`; + } + if (eventType === "job.cancelling") { + return `[${ts}] [WARN] Cancellation requested`; + } + if (eventType === "job.cancelled") { + return `[${ts}] [WARN] Job cancelled`; + } + if (eventType === "job.error") { + const error = + typeof event.payload.error === "string" && event.payload.error.length > 0 + ? event.payload.error + : "Job failed"; + return `[${ts}] [ERROR] ${error}`; + } + + return null; +} + +export function applyExecutionStatusSnapshot( + execution: RecipeExecutionRecord, + status: JobStatusResponse, +): RecipeExecutionRecord { + const mappedStatus = mapJobStatus(status.status); + return { + ...execution, + status: mappedStatus, + rows: status.rows ?? execution.rows, + stage: status.stage ?? execution.stage, + current_column: status.current_column ?? null, + progress: (normalizeObject(status.progress) as RecipeExecutionRecord["progress"]) ?? null, + column_progress: + (normalizeObject(status.column_progress) as RecipeExecutionRecord["column_progress"]) ?? + null, + model_usage: normalizeObject(status.model_usage), + artifact_path: status.artifact_path ?? execution.artifact_path, + error: status.error ?? null, + finishedAt: + mappedStatus === "completed" || + mappedStatus === "error" || + mappedStatus === "cancelled" + ? Date.now() + : null, + }; +} + +export function createBaseExecutionRecord(input: { + recipeId: string; + kind: RecipeExecutionKind; + rows: number; + currentSignature: string; +}): RecipeExecutionRecord { + const createdAt = Date.now(); + return { + id: crypto.randomUUID(), + recipeId: input.recipeId, + jobId: null, + kind: input.kind, + status: "pending", + rows: input.rows, + createdAt, + finishedAt: null, + recipeSignature: input.currentSignature, + stage: "pending", + current_column: null, + progress: null, + column_progress: null, + model_usage: null, + lastEventId: null, + artifact_path: null, + log_lines: [], + dataset: [], + datasetTotal: 0, + datasetPage: 1, + datasetPageSize: DATASET_PAGE_SIZE, + analysis: null, + processor_artifacts: null, + error: null, + }; +} diff --git a/studio/frontend/src/features/recipe-studio/executions/tracker.ts b/studio/frontend/src/features/recipe-studio/executions/tracker.ts new file mode 100644 index 0000000000..e150ad94be --- /dev/null +++ b/studio/frontend/src/features/recipe-studio/executions/tracker.ts @@ -0,0 +1,320 @@ +import { toastError, toastSuccess } from "@/shared/toast"; +import { + getRecipeJobAnalysis, + getRecipeJobDataset, + getRecipeJobStatus, + streamRecipeJobEvents, +} from "../api"; +import type { + RecipeExecutionKind, + RecipeExecutionProgress, + RecipeExecutionRecord, + RecipeExecutionStatus, +} from "../execution-types"; +import { + DATASET_PAGE_SIZE, + delay, + mapJobStatus, + normalizeAnalysis, + normalizeDatasetRows, + toErrorMessage, +} from "./execution-helpers"; +import { + appendExecutionLogLine, + applyExecutionStatusSnapshot, + toExecutionLogLine, +} from "./runtime"; + +type TrackRecipeExecutionParams = { + label: string; + kind: RecipeExecutionKind; + rows: number; + jobId: string; + initialExecution: RecipeExecutionRecord; + notify: boolean; + onUpsert: (record: RecipeExecutionRecord) => void; + onSetPreviewErrors: (errors: string[]) => void; + onPreviewSuccess?: () => void; +}; + +function isTerminalStatus(status: RecipeExecutionStatus): boolean { + return status === "completed" || status === "error" || status === "cancelled"; +} + +function normalizeCompletedProgress(input: { + latestExecution: RecipeExecutionRecord; + rows: number; +}): { + progress: RecipeExecutionProgress; + columnProgress: RecipeExecutionProgress | null; +} { + const { latestExecution, rows } = input; + const progressTotal = + typeof latestExecution.progress?.total === "number" && latestExecution.progress.total > 0 + ? latestExecution.progress.total + : latestExecution.rows > 0 + ? latestExecution.rows + : rows; + + const progress: RecipeExecutionProgress = { + ...(latestExecution.progress ?? {}), + done: progressTotal, + total: progressTotal, + percent: 100, + eta_sec: 0, + }; + + const columnProgress = + latestExecution.column_progress && + typeof latestExecution.column_progress.total === "number" && + latestExecution.column_progress.total > 0 + ? { + ...latestExecution.column_progress, + done: latestExecution.column_progress.total, + percent: 100, + eta_sec: 0, + } + : latestExecution.column_progress; + + return { progress, columnProgress }; +} + +export async function trackRecipeExecution({ + label, + kind, + rows, + jobId, + initialExecution, + notify, + onUpsert, + onSetPreviewErrors, + onPreviewSuccess, +}: TrackRecipeExecutionParams): Promise { + let done = false; + let lastStatus: RecipeExecutionStatus = initialExecution.status; + let completedEventPayload: Record | null = null; + let latestExecution: RecipeExecutionRecord = initialExecution; + + const eventsAbortController = new AbortController(); + void streamRecipeJobEvents({ + jobId, + signal: eventsAbortController.signal, + lastEventId: latestExecution.lastEventId, + onEvent: (event) => { + let changed = false; + + if (typeof event.id === "number") { + latestExecution = { + ...latestExecution, + lastEventId: event.id, + }; + changed = true; + } + + const logLine = toExecutionLogLine(event); + if (logLine) { + latestExecution = { + ...latestExecution, + log_lines: appendExecutionLogLine(latestExecution.log_lines, logLine), + }; + changed = true; + } + + const eventType = + typeof event.payload.type === "string" ? event.payload.type : event.event; + + if (eventType === "job.started") { + latestExecution = { + ...latestExecution, + status: "active", + }; + onUpsert(latestExecution); + return; + } + + if (eventType === "job.completed") { + lastStatus = "completed"; + completedEventPayload = event.payload; + done = true; + latestExecution = { + ...latestExecution, + status: "completed", + finishedAt: Date.now(), + artifact_path: + typeof event.payload.artifact_path === "string" + ? event.payload.artifact_path + : latestExecution.artifact_path, + error: null, + }; + onUpsert(latestExecution); + return; + } + + if (eventType === "job.error") { + lastStatus = "error"; + done = true; + latestExecution = { + ...latestExecution, + status: "error", + finishedAt: Date.now(), + error: + typeof event.payload.error === "string" + ? event.payload.error + : latestExecution.error ?? `${label} failed.`, + }; + onUpsert(latestExecution); + return; + } + + if (eventType === "job.cancelling") { + latestExecution = { + ...latestExecution, + status: "cancelling", + }; + onUpsert(latestExecution); + return; + } + + if (changed) { + onUpsert(latestExecution); + } + }, + }).catch(() => { + // polling is fallback source of truth + }); + + try { + while (!done) { + const status = await getRecipeJobStatus(jobId); + const mappedStatus = mapJobStatus(status.status); + lastStatus = mappedStatus; + latestExecution = applyExecutionStatusSnapshot(latestExecution, status); + onUpsert(latestExecution); + + done = isTerminalStatus(mappedStatus); + if (!done) { + await delay(1200); + } + } + } catch (error) { + const message = toErrorMessage(error, `${label} failed.`); + latestExecution = { + ...latestExecution, + status: "error", + error: message, + finishedAt: Date.now(), + }; + onUpsert(latestExecution); + if (notify) { + toastError(`${label} failed`, message); + } + return false; + } finally { + eventsAbortController.abort(); + } + + if (lastStatus === "completed") { + for (let attempt = 0; attempt < 3; attempt += 1) { + try { + const finalStatus = await getRecipeJobStatus(jobId); + latestExecution = applyExecutionStatusSnapshot(latestExecution, finalStatus); + } catch { + break; + } + if (attempt < 2) { + await delay(250); + } + } + + const eventAnalysis = completedEventPayload + ? completedEventPayload["analysis"] + : null; + const eventDataset = completedEventPayload + ? completedEventPayload["dataset"] + : null; + const shouldFetchPreviewDataset = kind === "preview" && !Array.isArray(eventDataset); + const shouldFetchAnalysis = + !completedEventPayload || + typeof eventAnalysis !== "object" || + eventAnalysis === null || + kind === "full"; + + const [analysisResult, datasetResult] = await Promise.allSettled([ + shouldFetchAnalysis + ? getRecipeJobAnalysis(jobId) + : Promise.resolve(eventAnalysis), + shouldFetchPreviewDataset || kind === "full" + ? getRecipeJobDataset(jobId, { limit: DATASET_PAGE_SIZE, offset: 0 }) + : Promise.resolve({ dataset: eventDataset ?? [], total: rows }), + ]); + + const analysis = + analysisResult.status === "fulfilled" + ? normalizeAnalysis(analysisResult.value) + : latestExecution.analysis; + const datasetResponse = + datasetResult.status === "fulfilled" + ? datasetResult.value + : null; + const dataset = datasetResponse + ? normalizeDatasetRows(datasetResponse.dataset) + : latestExecution.dataset; + const datasetTotal = + datasetResponse && typeof datasetResponse.total === "number" + ? datasetResponse.total + : latestExecution.datasetTotal; + const completedProgress = normalizeCompletedProgress({ latestExecution, rows }); + + latestExecution = { + ...latestExecution, + status: "completed", + progress: completedProgress.progress, + column_progress: completedProgress.columnProgress, + analysis, + dataset, + datasetTotal, + datasetPage: 1, + datasetPageSize: DATASET_PAGE_SIZE, + error: null, + finishedAt: latestExecution.finishedAt ?? Date.now(), + }; + onUpsert(latestExecution); + + if (notify) { + if (kind === "preview") { + onSetPreviewErrors([]); + onPreviewSuccess?.(); + toastSuccess(`Preview generated (${rows} rows).`); + } else { + toastSuccess("Full run completed."); + } + } + return true; + } + + if (lastStatus === "cancelled") { + latestExecution = { + ...latestExecution, + status: "cancelled", + error: latestExecution.error ?? "Run cancelled.", + finishedAt: latestExecution.finishedAt ?? Date.now(), + }; + onUpsert(latestExecution); + if (notify) { + toastError(`${label} cancelled`, "The execution was cancelled."); + } + return false; + } + + latestExecution = { + ...latestExecution, + status: "error", + error: latestExecution.error ?? `${label} failed.`, + finishedAt: latestExecution.finishedAt ?? Date.now(), + }; + onUpsert(latestExecution); + if (notify) { + toastError(`${label} failed`, latestExecution.error ?? "Execution failed."); + } + return false; +} diff --git a/studio/frontend/src/features/recipe-studio/hooks/use-recipe-executions.ts b/studio/frontend/src/features/recipe-studio/hooks/use-recipe-executions.ts index 3f79a2ce32..d00c9955e5 100644 --- a/studio/frontend/src/features/recipe-studio/hooks/use-recipe-executions.ts +++ b/studio/frontend/src/features/recipe-studio/hooks/use-recipe-executions.ts @@ -1,35 +1,27 @@ import { useCallback, useEffect } from "react"; import { useShallow } from "zustand/react/shallow"; -import { toastError, toastSuccess } from "@/shared/toast"; +import { toastError } from "@/shared/toast"; import { cancelRecipeJob, createRecipeJob, - getRecipeJobAnalysis, getRecipeJobDataset, - getRecipeJobStatus, - streamRecipeJobEvents, validateRecipe, - type JobEvent, - type JobStatusResponse, } from "../api"; -import { listRecipeExecutions, saveRecipeExecution } from "../data/executions-db"; -import type { - RecipeExecutionRecord, - RecipeExecutionStatus, -} from "../execution-types"; +import { saveRecipeExecution } from "../data/executions-db"; +import type { RecipeExecutionRecord } from "../execution-types"; import { DATASET_PAGE_SIZE, - delay, executionLabel, - isExecutionInProgress, - mapJobStatus, - normalizeAnalysis, normalizeDatasetRows, - normalizeObject, - sortExecutions, toErrorMessage, withExecutionDefaults, } from "../executions/execution-helpers"; +import { + findResumableExecution, + loadSortedRecipeExecutions, +} from "../executions/hydration"; +import { createBaseExecutionRecord } from "../executions/runtime"; +import { trackRecipeExecution } from "../executions/tracker"; import { useRecipeExecutionsStore } from "../stores/recipe-executions"; import type { RecipePayload, RecipePayloadResult } from "../utils/payload/types"; @@ -59,127 +51,6 @@ type UseRecipeExecutionsResult = { loadExecutionDatasetPage: (id: string, page: number) => Promise; }; -const MAX_LOG_LINES = 1500; - -function formatEventTime(ts: unknown): string { - if (typeof ts !== "number" || !Number.isFinite(ts)) { - return new Date().toLocaleTimeString(); - } - const ms = ts > 10_000_000_000 ? ts : ts * 1000; - return new Date(ms).toLocaleTimeString(); -} - -function appendLogLine(lines: string[], nextLine: string): string[] { - const next = [...lines, nextLine]; - if (next.length <= MAX_LOG_LINES) { - return next; - } - return next.slice(next.length - MAX_LOG_LINES); -} - -function toLogLine(event: JobEvent): string | null { - const eventType = - typeof event.payload.type === "string" ? event.payload.type : event.event; - const ts = formatEventTime(event.payload.ts); - - if (eventType === "log") { - const message = - typeof event.payload.message === "string" ? event.payload.message.trim() : ""; - if (!message) { - return null; - } - const level = - typeof event.payload.level === "string" && event.payload.level.length > 0 - ? event.payload.level.toUpperCase() - : "INFO"; - return `[${ts}] [${level}] ${message}`; - } - - if (eventType === "job.started") { - return `[${ts}] [INFO] Job started`; - } - if (eventType === "job.completed") { - return `[${ts}] [INFO] Job completed`; - } - if (eventType === "job.cancelling") { - return `[${ts}] [WARN] Cancellation requested`; - } - if (eventType === "job.cancelled") { - return `[${ts}] [WARN] Job cancelled`; - } - if (eventType === "job.error") { - const error = - typeof event.payload.error === "string" && event.payload.error.length > 0 - ? event.payload.error - : "Job failed"; - return `[${ts}] [ERROR] ${error}`; - } - - return null; -} - -function applyStatusSnapshot( - execution: RecipeExecutionRecord, - status: JobStatusResponse, -): RecipeExecutionRecord { - const mappedStatus = mapJobStatus(status.status); - return { - ...execution, - status: mappedStatus, - rows: status.rows ?? execution.rows, - stage: status.stage ?? execution.stage, - current_column: status.current_column ?? null, - progress: (normalizeObject(status.progress) as RecipeExecutionRecord["progress"]) ?? null, - column_progress: - (normalizeObject(status.column_progress) as RecipeExecutionRecord["column_progress"]) ?? - null, - model_usage: normalizeObject(status.model_usage), - artifact_path: status.artifact_path ?? execution.artifact_path, - error: status.error ?? null, - finishedAt: - mappedStatus === "completed" || - mappedStatus === "error" || - mappedStatus === "cancelled" - ? Date.now() - : null, - }; -} - -function createBaseExecution(input: { - recipeId: string; - kind: "preview" | "full"; - rows: number; - currentSignature: string; -}): RecipeExecutionRecord { - const createdAt = Date.now(); - return { - id: crypto.randomUUID(), - recipeId: input.recipeId, - jobId: null, - kind: input.kind, - status: "pending", - rows: input.rows, - createdAt, - finishedAt: null, - recipeSignature: input.currentSignature, - stage: "pending", - current_column: null, - progress: null, - column_progress: null, - model_usage: null, - lastEventId: null, - artifact_path: null, - log_lines: [], - dataset: [], - datasetTotal: 0, - datasetPage: 1, - datasetPageSize: DATASET_PAGE_SIZE, - analysis: null, - processor_artifacts: null, - error: null, - }; -} - export function useRecipeExecutions({ recipeId, currentSignature, @@ -224,328 +95,66 @@ export function useRecipeExecutions({ resetForRecipe: state.resetForRecipe, })), ); + const payloadErrorMessage = payloadResult.errors[0] ?? "Invalid payload."; const upsertAndPersist = useCallback( (record: RecipeExecutionRecord): void => { - const normalized = withExecutionDefaults(record); - upsertExecution(normalized); - void saveRecipeExecution(normalized).catch((error) => { + const normalizedRecord = withExecutionDefaults(record); + upsertExecution(normalizedRecord); + void saveRecipeExecution(normalizedRecord).catch((error) => { console.error("Save recipe execution failed:", error); }); }, [upsertExecution], ); - const trackExecution = useCallback( - async (input: { - label: string; - kind: "preview" | "full"; - rows: number; - jobId: string; - initialExecution: RecipeExecutionRecord; - notify: boolean; - }): Promise => { - const { label, kind, rows, jobId, notify } = input; - let done = false; - let lastStatus: RecipeExecutionStatus = input.initialExecution.status; - let completedEventPayload: Record | null = null; - let latestExecution: RecipeExecutionRecord = input.initialExecution; - - const eventsAbortController = new AbortController(); - void streamRecipeJobEvents({ - jobId, - signal: eventsAbortController.signal, - lastEventId: latestExecution.lastEventId, - onEvent: (event) => { - let changed = false; - - if (typeof event.id === "number") { - latestExecution = { - ...latestExecution, - lastEventId: event.id, - }; - changed = true; - } - - const logLine = toLogLine(event); - if (logLine) { - latestExecution = { - ...latestExecution, - log_lines: appendLogLine(latestExecution.log_lines, logLine), - }; - changed = true; - } - - const eventType = - typeof event.payload.type === "string" ? event.payload.type : event.event; - - if (eventType === "job.started") { - latestExecution = { - ...latestExecution, - status: "active", - }; - upsertAndPersist(latestExecution); - return; - } - - if (eventType === "job.completed") { - lastStatus = "completed"; - completedEventPayload = event.payload; - done = true; - latestExecution = { - ...latestExecution, - status: "completed", - finishedAt: Date.now(), - artifact_path: - typeof event.payload.artifact_path === "string" - ? event.payload.artifact_path - : latestExecution.artifact_path, - error: null, - }; - upsertAndPersist(latestExecution); - return; - } - - if (eventType === "job.error") { - lastStatus = "error"; - done = true; - latestExecution = { - ...latestExecution, - status: "error", - finishedAt: Date.now(), - error: - typeof event.payload.error === "string" - ? event.payload.error - : latestExecution.error ?? `${label} failed.`, - }; - upsertAndPersist(latestExecution); - return; - } - - if (eventType === "job.cancelling") { - latestExecution = { - ...latestExecution, - status: "cancelling", - }; - upsertAndPersist(latestExecution); - return; - } - - if (changed) { - upsertAndPersist(latestExecution); - } - }, - }).catch(() => { - // polling is fallback source of truth - }); - - try { - while (!done) { - const status = await getRecipeJobStatus(jobId); - const mappedStatus = mapJobStatus(status.status); - lastStatus = mappedStatus; - latestExecution = applyStatusSnapshot(latestExecution, status); - upsertAndPersist(latestExecution); - - done = - mappedStatus === "completed" || - mappedStatus === "error" || - mappedStatus === "cancelled"; - if (!done) { - await delay(1200); - } - } - } catch (error) { - const message = toErrorMessage(error, `${label} failed.`); - latestExecution = { - ...latestExecution, - status: "error", - error: message, - finishedAt: Date.now(), - }; - upsertAndPersist(latestExecution); - if (notify) { - toastError(`${label} failed`, message); - } - return false; - } finally { - eventsAbortController.abort(); - } - - if (lastStatus === "completed") { - for (let attempt = 0; attempt < 3; attempt += 1) { - try { - const finalStatus = await getRecipeJobStatus(jobId); - latestExecution = applyStatusSnapshot(latestExecution, finalStatus); - } catch { - break; - } - if (attempt < 2) { - await delay(250); - } - } - - const eventAnalysis = completedEventPayload - ? completedEventPayload["analysis"] - : null; - const eventDataset = completedEventPayload - ? completedEventPayload["dataset"] - : null; - const shouldFetchPreviewDataset = - kind === "preview" && !Array.isArray(eventDataset); - const shouldFetchAnalysis = - !completedEventPayload || - typeof eventAnalysis !== "object" || - eventAnalysis === null || - kind === "full"; - - const [analysisResult, datasetResult] = await Promise.allSettled([ - shouldFetchAnalysis - ? getRecipeJobAnalysis(jobId) - : Promise.resolve(eventAnalysis), - shouldFetchPreviewDataset || kind === "full" - ? getRecipeJobDataset(jobId, { limit: DATASET_PAGE_SIZE, offset: 0 }) - : Promise.resolve({ dataset: eventDataset ?? [], total: rows }), - ]); - - const analysis = - analysisResult.status === "fulfilled" - ? normalizeAnalysis(analysisResult.value) - : latestExecution.analysis; - const datasetResponse = - datasetResult.status === "fulfilled" - ? datasetResult.value - : null; - const dataset = datasetResponse - ? normalizeDatasetRows(datasetResponse.dataset) - : latestExecution.dataset; - const datasetTotal = - datasetResponse && typeof datasetResponse.total === "number" - ? datasetResponse.total - : latestExecution.datasetTotal; - - const progressTotal = - typeof latestExecution.progress?.total === "number" && - latestExecution.progress.total > 0 - ? latestExecution.progress.total - : latestExecution.rows > 0 - ? latestExecution.rows - : rows; - - const completedProgress: RecipeExecutionRecord["progress"] = { - ...(latestExecution.progress ?? {}), - done: progressTotal, - total: progressTotal, - percent: 100, - eta_sec: 0, - }; - const completedColumnProgress: RecipeExecutionRecord["column_progress"] = - latestExecution.column_progress && - typeof latestExecution.column_progress.total === "number" && - latestExecution.column_progress.total > 0 - ? { - ...latestExecution.column_progress, - done: latestExecution.column_progress.total, - percent: 100, - eta_sec: 0, - } - : latestExecution.column_progress; - - latestExecution = { - ...latestExecution, - status: "completed", - progress: completedProgress, - column_progress: completedColumnProgress, - analysis, - dataset, - datasetTotal, - datasetPage: 1, - datasetPageSize: DATASET_PAGE_SIZE, - error: null, - finishedAt: latestExecution.finishedAt ?? Date.now(), - }; - upsertAndPersist(latestExecution); - - if (notify) { - if (kind === "preview") { - setPreviewErrors([]); - onPreviewSuccess?.(); - toastSuccess(`Preview generated (${rows} rows).`); - } else { - toastSuccess("Full run completed."); - } - } - return true; - } - - if (lastStatus === "cancelled") { - latestExecution = { - ...latestExecution, - status: "cancelled", - error: latestExecution.error ?? "Run cancelled.", - finishedAt: latestExecution.finishedAt ?? Date.now(), - }; - upsertAndPersist(latestExecution); - if (notify) { - toastError(`${label} cancelled`, "The execution was cancelled."); - } - return false; - } - - latestExecution = { - ...latestExecution, - status: "error", - error: latestExecution.error ?? `${label} failed.`, - finishedAt: latestExecution.finishedAt ?? Date.now(), - }; - upsertAndPersist(latestExecution); - if (notify) { - toastError(`${label} failed`, latestExecution.error ?? "Execution failed."); - } - return false; - }, - [onPreviewSuccess, setPreviewErrors, upsertAndPersist], - ); - useEffect(() => { let cancelled = false; resetForRecipe(); - async function loadExecutions(): Promise { + async function hydrate(): Promise { try { - const records = await listRecipeExecutions(recipeId); + const records = await loadSortedRecipeExecutions(recipeId); if (cancelled) { return; } - const sortedRecords = sortExecutions(records.map(withExecutionDefaults)); - setExecutions(sortedRecords); - const activeExecution = sortedRecords.find( - (record) => record.jobId && isExecutionInProgress(record.status), - ); - - if (activeExecution?.jobId) { - void trackExecution({ - label: executionLabel(activeExecution.kind), - kind: activeExecution.kind, - rows: activeExecution.rows, - jobId: activeExecution.jobId, - initialExecution: activeExecution, - notify: false, - }); + setExecutions(records); + const resumable = findResumableExecution(records); + if (!resumable?.jobId) { + return; } + + void trackRecipeExecution({ + label: executionLabel(resumable.kind), + kind: resumable.kind, + rows: resumable.rows, + jobId: resumable.jobId, + initialExecution: resumable, + notify: false, + onUpsert: upsertAndPersist, + onSetPreviewErrors: setPreviewErrors, + onPreviewSuccess, + }); } catch (error) { console.error("Load recipe executions failed:", error); } } - void loadExecutions(); + void hydrate(); return () => { cancelled = true; }; - }, [recipeId, resetForRecipe, setExecutions, trackExecution]); + }, [ + onPreviewSuccess, + recipeId, + resetForRecipe, + setExecutions, + setPreviewErrors, + upsertAndPersist, + ]); const readPayload = useCallback((): RecipePayload | null => { if (payloadResult.errors.length === 0) { @@ -553,6 +162,16 @@ export function useRecipeExecutions({ } return null; }, [payloadResult.errors.length, payloadResult.payload]); + const readExecutablePayload = useCallback((): RecipePayload | null => { + const payload = readPayload(); + if (payload) { + return payload; + } + + setPreviewErrors(payloadResult.errors); + toastError("Invalid recipe payload", payloadErrorMessage); + return null; + }, [payloadErrorMessage, payloadResult.errors, readPayload, setPreviewErrors]); const openPreviewDialog = useCallback((): void => { setPreviewErrors([]); @@ -570,7 +189,7 @@ export function useRecipeExecutions({ const label = executionLabel(kind); setLoading(true); - const baseExecution = createBaseExecution({ + const baseExecution = createBaseExecutionRecord({ recipeId, kind, rows, @@ -594,19 +213,22 @@ export function useRecipeExecutions({ }, }; const createdJob = await createRecipeJob(jobPayload); - const latestExecution = { + const executionWithJob = { ...baseExecution, jobId: createdJob.job_id, }; - upsertAndPersist(latestExecution); + upsertAndPersist(executionWithJob); - return await trackExecution({ + return await trackRecipeExecution({ label, kind, rows, jobId: createdJob.job_id, - initialExecution: latestExecution, + initialExecution: executionWithJob, notify: true, + onUpsert: upsertAndPersist, + onSetPreviewErrors: setPreviewErrors, + onPreviewSuccess, }); } catch (error) { const message = toErrorMessage(error, `${label} request failed.`); @@ -628,22 +250,19 @@ export function useRecipeExecutions({ [ currentSignature, onExecutionStart, + onPreviewSuccess, recipeId, setFullLoading, setPreviewDialogOpen, setPreviewErrors, setPreviewLoading, - trackExecution, upsertAndPersist, ], ); const runPreview = useCallback(async (): Promise => { - const payload = readPayload(); - const payloadErrorMessage = payloadResult.errors[0] ?? "Invalid payload."; + const payload = readExecutablePayload(); if (!payload) { - setPreviewErrors(payloadResult.errors); - toastError("Invalid recipe payload", payloadErrorMessage); return false; } @@ -676,14 +295,11 @@ export function useRecipeExecutions({ payload, rows: previewRows, }); - }, [payloadResult.errors, previewRows, readPayload, runExecution, setPreviewErrors]); + }, [previewRows, readExecutablePayload, runExecution, setPreviewErrors]); const runFull = useCallback(async (): Promise => { - const payload = readPayload(); - const payloadErrorMessage = payloadResult.errors[0] ?? "Invalid payload."; + const payload = readExecutablePayload(); if (!payload) { - setPreviewErrors(payloadResult.errors); - toastError("Invalid recipe payload", payloadErrorMessage); return false; } @@ -698,7 +314,7 @@ export function useRecipeExecutions({ payload, rows, }); - }, [payloadResult.errors, readPayload, runExecution, setPreviewErrors]); + }, [readExecutablePayload, runExecution]); const cancelExecution = useCallback( async (id: string): Promise => { @@ -723,10 +339,7 @@ export function useRecipeExecutions({ const loadExecutionDatasetPage = useCallback( async (id: string, page: number): Promise => { const execution = executions.find((entry) => entry.id === id); - if (!execution || execution.kind !== "full" || !execution.jobId) { - return; - } - if (page < 1) { + if (!execution || execution.kind !== "full" || !execution.jobId || page < 1) { return; }