1225 lines
38 KiB
TypeScript
1225 lines
38 KiB
TypeScript
import { create } from "zustand";
|
||
|
||
import {
|
||
ApiRequestError,
|
||
type CronPreview,
|
||
type Schedule,
|
||
type ScheduleArtifact,
|
||
type ScheduleEdge,
|
||
type ScheduleNode,
|
||
type ScheduleRunDetail,
|
||
type ScheduleRunSummary,
|
||
type WorkspaceBoundApi,
|
||
} from "../../../services/api";
|
||
|
||
import { useScriptWorkspaceStore } from "../../platform/state/scriptWorkspaceStore";
|
||
import { useUiStore } from "../../platform/state/uiStore";
|
||
|
||
type Notice = {
|
||
tone: "success" | "error" | "info";
|
||
message: string;
|
||
};
|
||
|
||
// ---- Types ----
|
||
|
||
export type ScheduleForm = {
|
||
scheduleName: string;
|
||
description: string;
|
||
triggerType: "manual" | "cron" | "api";
|
||
cronExpression: string;
|
||
timezone: string;
|
||
enabled: boolean;
|
||
maxConcurrency: string;
|
||
failurePolicy: "stop" | "continue";
|
||
};
|
||
|
||
export type PythonVersion = "3.8" | "3.10" | "3.12";
|
||
|
||
export const PYTHON_VERSION_OPTIONS: readonly PythonVersion[] = [
|
||
"3.8",
|
||
"3.10",
|
||
"3.12",
|
||
];
|
||
|
||
export type NodeForm = {
|
||
nodeName: string;
|
||
timeoutSeconds: string;
|
||
retryCount: string;
|
||
retryIntervalSec: string;
|
||
argumentsJson: string;
|
||
envRefsJson: string;
|
||
pythonVersion: PythonVersion;
|
||
};
|
||
|
||
export type NodePositionDraft = {
|
||
position_x: number;
|
||
position_y: number;
|
||
};
|
||
|
||
export type ScheduleContextMenu =
|
||
| { x: number; y: number; kind: "schedule-list" }
|
||
| { x: number; y: number; kind: "schedule"; schedule: Schedule }
|
||
| {
|
||
x: number;
|
||
y: number;
|
||
kind: "artifact";
|
||
artifact: ScheduleArtifact;
|
||
}
|
||
| { x: number; y: number; kind: "node"; node: ScheduleNode }
|
||
| { x: number; y: number; kind: "edge"; edge: ScheduleEdge };
|
||
|
||
export const EMPTY_SCHEDULE_FORM: ScheduleForm = {
|
||
scheduleName: "",
|
||
description: "",
|
||
triggerType: "manual",
|
||
cronExpression: "",
|
||
timezone: "Asia/Shanghai",
|
||
enabled: false,
|
||
maxConcurrency: "1",
|
||
failurePolicy: "stop",
|
||
};
|
||
|
||
export const EMPTY_NODE_FORM: NodeForm = {
|
||
nodeName: "",
|
||
timeoutSeconds: "600",
|
||
retryCount: "0",
|
||
retryIntervalSec: "5",
|
||
argumentsJson: "{}",
|
||
envRefsJson: "{}",
|
||
pythonVersion: "3.12",
|
||
};
|
||
|
||
// ---- Module-level non-reactive holders ----
|
||
|
||
let _api: WorkspaceBoundApi | null = null;
|
||
const positionDrafts = new Map<string, NodePositionDraft>();
|
||
|
||
export const bindSchedulesApi = (api: WorkspaceBoundApi | null) => {
|
||
_api = api;
|
||
};
|
||
|
||
function requireApi(): WorkspaceBoundApi {
|
||
if (!_api) throw new Error("schedules API 未绑定");
|
||
return _api;
|
||
}
|
||
|
||
function notify(notice: Notice) {
|
||
useUiStore.getState().pushToast(notice);
|
||
}
|
||
|
||
function setApiOnline(online: boolean) {
|
||
useScriptWorkspaceStore.getState().setApiOnline(online);
|
||
}
|
||
|
||
// ---- Helpers ----
|
||
|
||
function scheduleToForm(schedule: Schedule): ScheduleForm {
|
||
return {
|
||
scheduleName: schedule.schedule_name,
|
||
description: schedule.description ?? "",
|
||
triggerType: schedule.trigger_type,
|
||
cronExpression: schedule.cron_expression ?? "",
|
||
timezone: schedule.timezone,
|
||
enabled: schedule.enabled,
|
||
maxConcurrency: String(schedule.max_concurrency),
|
||
failurePolicy: schedule.failure_policy,
|
||
};
|
||
}
|
||
|
||
function nodeToForm(node: ScheduleNode): NodeForm {
|
||
return {
|
||
nodeName: node.node_name,
|
||
timeoutSeconds: String(node.timeout_seconds),
|
||
retryCount: String(node.retry_count),
|
||
retryIntervalSec: String(node.retry_interval_sec),
|
||
argumentsJson: JSON.stringify(node.arguments_json ?? {}, null, 2),
|
||
envRefsJson: JSON.stringify(node.env_refs_json ?? {}, null, 2),
|
||
pythonVersion: (node.python_version ?? "3.12") as PythonVersion,
|
||
};
|
||
}
|
||
|
||
function parseObject(text: string, label: string): Record<string, unknown> {
|
||
const trimmed = text.trim();
|
||
if (!trimmed) return {};
|
||
try {
|
||
const parsed = JSON.parse(trimmed);
|
||
if (parsed === null || typeof parsed !== "object" || Array.isArray(parsed)) {
|
||
throw new Error(`${label} 必须是 JSON 对象`);
|
||
}
|
||
return parsed as Record<string, unknown>;
|
||
} catch (error) {
|
||
if (error instanceof SyntaxError) {
|
||
throw new Error(`${label} JSON 格式不正确`);
|
||
}
|
||
throw error;
|
||
}
|
||
}
|
||
|
||
function applyPositionDrafts(schedule: Schedule): Schedule {
|
||
if (positionDrafts.size === 0) return schedule;
|
||
return {
|
||
...schedule,
|
||
nodes: schedule.nodes.map((node) => {
|
||
const draft = positionDrafts.get(node.node_id);
|
||
return draft ? { ...node, ...draft } : node;
|
||
}),
|
||
};
|
||
}
|
||
|
||
function clearPositionDrafts() {
|
||
positionDrafts.clear();
|
||
}
|
||
|
||
// ---- Store ----
|
||
|
||
type LogState = {
|
||
loading: boolean;
|
||
content: string | null;
|
||
fileName: string | null;
|
||
error: string | null;
|
||
};
|
||
|
||
type State = {
|
||
schedules: Schedule[];
|
||
artifacts: ScheduleArtifact[];
|
||
schedule: Schedule | null;
|
||
|
||
loading: boolean;
|
||
busy: string | null;
|
||
|
||
selectedNodeId: string | null;
|
||
selectedEdgeId: string | null;
|
||
linkSourceId: string | null;
|
||
|
||
scheduleKeyword: string;
|
||
artifactKeyword: string;
|
||
|
||
createDialogOpen: boolean;
|
||
newScheduleName: string;
|
||
|
||
scheduleForm: ScheduleForm;
|
||
nodeForm: NodeForm;
|
||
cronResult: CronPreview | null;
|
||
contextMenu: ScheduleContextMenu | null;
|
||
|
||
runs: ScheduleRunSummary[];
|
||
runsLoading: boolean;
|
||
|
||
expandedRunId: string | null;
|
||
runDetails: Record<string, ScheduleRunDetail>;
|
||
detailLoadingId: string | null;
|
||
detailErrors: Record<string, string>;
|
||
openLogNodeRunIds: Set<string>;
|
||
logStates: Record<string, LogState>;
|
||
artifactBusyKey: string | null;
|
||
artifactErrors: Record<string, string>;
|
||
positionDraftCount: number;
|
||
};
|
||
|
||
type Actions = {
|
||
bindApi: (api: WorkspaceBoundApi | null) => void;
|
||
reset: () => void;
|
||
|
||
// pure setters (mostly used by canvas/drag/inspector)
|
||
setSchedule: (
|
||
value: Schedule | null | ((current: Schedule | null) => Schedule | null),
|
||
) => void;
|
||
setSchedules: (
|
||
value: Schedule[] | ((current: Schedule[]) => Schedule[]),
|
||
) => void;
|
||
setArtifacts: (
|
||
value: ScheduleArtifact[]
|
||
| ((current: ScheduleArtifact[]) => ScheduleArtifact[]),
|
||
) => void;
|
||
setLoading: (loading: boolean) => void;
|
||
setBusy: (busy: string | null) => void;
|
||
setRuns: (
|
||
value: ScheduleRunSummary[]
|
||
| ((current: ScheduleRunSummary[]) => ScheduleRunSummary[]),
|
||
) => void;
|
||
setRunsLoading: (loading: boolean) => void;
|
||
setPositionDraftCount: (count: number) => void;
|
||
setRunDetails: (
|
||
updater: (current: Record<string, ScheduleRunDetail>) => Record<string, ScheduleRunDetail>,
|
||
) => void;
|
||
setOpenLogNodeRunIds: (
|
||
value: Set<string> | ((current: Set<string>) => Set<string>),
|
||
) => void;
|
||
setLogStates: (
|
||
updater: (current: Record<string, {
|
||
loading: boolean;
|
||
content: string | null;
|
||
fileName: string | null;
|
||
error: string | null;
|
||
}>) => Record<string, {
|
||
loading: boolean;
|
||
content: string | null;
|
||
fileName: string | null;
|
||
error: string | null;
|
||
}>,
|
||
) => void;
|
||
setArtifactBusyKey: (key: string | null) => void;
|
||
setArtifactErrors: (
|
||
updater: (current: Record<string, string>) => Record<string, string>,
|
||
) => void;
|
||
setDetailLoadingId: (id: string | null) => void;
|
||
setDetailErrors: (
|
||
updater: (current: Record<string, string>) => Record<string, string>,
|
||
) => void;
|
||
setSelectedNodeId: (id: string | null) => void;
|
||
setSelectedEdgeId: (id: string | null) => void;
|
||
setLinkSourceId: (id: string | null) => void;
|
||
setScheduleKeyword: (k: string) => void;
|
||
setArtifactKeyword: (k: string) => void;
|
||
setCreateDialogOpen: (open: boolean) => void;
|
||
setNewScheduleName: (name: string) => void;
|
||
setScheduleForm: (form: ScheduleForm) => void;
|
||
setNodeForm: (form: NodeForm) => void;
|
||
setCronResult: (result: CronPreview | null) => void;
|
||
setContextMenu: (menu: ScheduleContextMenu | null) => void;
|
||
closeContextMenu: () => void;
|
||
|
||
// data loaders
|
||
loadInitial: () => Promise<void>;
|
||
refreshLists: (preferredScheduleId?: string | null) => Promise<void>;
|
||
refreshRuns: (scheduleId: string, showLoading?: boolean) => Promise<void>;
|
||
chooseSchedule: (scheduleId: string) => Promise<void>;
|
||
|
||
// mutations
|
||
openCreateDialog: () => void;
|
||
addSchedule: (name: string) => Promise<void>;
|
||
removeSchedule: (target?: Schedule) => Promise<void>;
|
||
renameSchedule: (target: Schedule) => Promise<void>;
|
||
removeArtifact: (artifact: ScheduleArtifact) => Promise<void>;
|
||
saveSchedule: () => Promise<void>;
|
||
runCronPreview: () => Promise<void>;
|
||
runNow: () => Promise<void>;
|
||
addArtifactAt: (
|
||
artifact: ScheduleArtifact,
|
||
positionX: number,
|
||
positionY: number,
|
||
) => Promise<void>;
|
||
saveNode: (selectedNode: ScheduleNode | null) => Promise<void>;
|
||
removeNode: (target?: ScheduleNode) => Promise<void>;
|
||
removeEdge: (target?: ScheduleEdge) => Promise<void>;
|
||
checkDag: () => Promise<void>;
|
||
connectTo: (targetNodeId: string) => Promise<void>;
|
||
|
||
// runs / log helpers
|
||
toggleRun: (runId: string) => void;
|
||
toggleLog: (runId: string, nodeRunId: string) => Promise<void>;
|
||
downloadResult: (runId: string, nodeRunId: string) => Promise<void>;
|
||
downloadLog: (runId: string, nodeRunId: string) => Promise<void>;
|
||
setExpandedRunId: (
|
||
id: string | null | ((current: string | null) => string | null),
|
||
) => void;
|
||
|
||
// drag helpers
|
||
setPositionDraft: (nodeId: string, draft: NodePositionDraft) => void;
|
||
getPositionDrafts: () => Map<string, NodePositionDraft>;
|
||
clearAllPositionDrafts: () => void;
|
||
prunePositionDrafts: (validNodeIds: Set<string>) => void;
|
||
};
|
||
|
||
const initial: State = {
|
||
schedules: [],
|
||
artifacts: [],
|
||
schedule: null,
|
||
loading: true,
|
||
busy: null,
|
||
selectedNodeId: null,
|
||
selectedEdgeId: null,
|
||
linkSourceId: null,
|
||
scheduleKeyword: "",
|
||
artifactKeyword: "",
|
||
createDialogOpen: false,
|
||
newScheduleName: "",
|
||
scheduleForm: EMPTY_SCHEDULE_FORM,
|
||
nodeForm: EMPTY_NODE_FORM,
|
||
cronResult: null,
|
||
contextMenu: null,
|
||
runs: [],
|
||
runsLoading: false,
|
||
expandedRunId: null,
|
||
runDetails: {},
|
||
detailLoadingId: null,
|
||
detailErrors: {},
|
||
openLogNodeRunIds: new Set(),
|
||
logStates: {},
|
||
artifactBusyKey: null,
|
||
artifactErrors: {},
|
||
positionDraftCount: 0,
|
||
};
|
||
|
||
export const useSchedulesStore = create<State & Actions>((set, get) => {
|
||
// Generic error handler with 412 (concurrent update) awareness
|
||
async function handleError(error: unknown, fallback: string): Promise<void> {
|
||
if (error instanceof ApiRequestError && error.status === 412) {
|
||
const currentId = get().schedule?.schedule_id;
|
||
if (currentId) {
|
||
get().refreshLists(currentId).catch(() => undefined);
|
||
}
|
||
notify({ tone: "error", message: "调度已被其他操作更新，已重新加载最新版本" });
|
||
return;
|
||
}
|
||
notify({
|
||
tone: "error",
|
||
message: error instanceof Error ? error.message : fallback,
|
||
});
|
||
}
|
||
|
||
// Generic mutation wrapper: busy + try/catch + 412 + positionDraft cleanup + schedules list update
|
||
async function withMutation(
|
||
label: string,
|
||
action: () => Promise<Schedule>,
|
||
successMessage: string,
|
||
): Promise<Schedule | null> {
|
||
const api = requireApi();
|
||
if (get().busy) return null;
|
||
set({ busy: label });
|
||
try {
|
||
const serverUpdated = await action();
|
||
const validNodeIds = new Set(
|
||
serverUpdated.nodes.map((node) => node.node_id),
|
||
);
|
||
for (const nodeId of [...positionDrafts.keys()]) {
|
||
if (!validNodeIds.has(nodeId)) positionDrafts.delete(nodeId);
|
||
}
|
||
const updated = applyPositionDrafts(serverUpdated);
|
||
set((state) => ({
|
||
schedule: updated,
|
||
schedules: (() => {
|
||
const summary = { ...updated, nodes: [], edges: [] };
|
||
const index = state.schedules.findIndex(
|
||
(item) => item.schedule_id === updated.schedule_id,
|
||
);
|
||
if (index < 0) return [summary, ...state.schedules];
|
||
return state.schedules.map((item) =>
|
||
item.schedule_id === updated.schedule_id ? summary : item
|
||
);
|
||
})(),
|
||
positionDraftCount: positionDrafts.size,
|
||
}));
|
||
notify({ tone: "success", message: successMessage });
|
||
return updated;
|
||
} catch (error) {
|
||
await handleError(error, `${successMessage}失败`);
|
||
return null;
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
}
|
||
|
||
return {
|
||
...initial,
|
||
|
||
bindApi: bindSchedulesApi,
|
||
|
||
reset: () => {
|
||
set({ ...initial, loading: true });
|
||
},
|
||
|
||
|
||
// ---- pure setters ----
|
||
|
||
setSchedule: (value) =>
|
||
set((state) => ({
|
||
schedule:
|
||
typeof value === "function"
|
||
? (value as (current: Schedule | null) => Schedule | null)(state.schedule)
|
||
: value,
|
||
})),
|
||
setSchedules: (value) =>
|
||
set((state) => ({
|
||
schedules:
|
||
typeof value === "function"
|
||
? (value as (current: Schedule[]) => Schedule[])(state.schedules)
|
||
: value,
|
||
})),
|
||
setArtifacts: (value) =>
|
||
set((state) => ({
|
||
artifacts:
|
||
typeof value === "function"
|
||
? (value as (current: ScheduleArtifact[]) => ScheduleArtifact[])(
|
||
state.artifacts,
|
||
)
|
||
: value,
|
||
})),
|
||
setLoading: (loading) => set({ loading }),
|
||
setBusy: (busy) => set({ busy }),
|
||
setRuns: (value) =>
|
||
set((state) => ({
|
||
runs:
|
||
typeof value === "function"
|
||
? (value as (current: ScheduleRunSummary[]) => ScheduleRunSummary[])(
|
||
state.runs,
|
||
)
|
||
: value,
|
||
})),
|
||
setRunsLoading: (loading) => set({ runsLoading: loading }),
|
||
setRunDetails: (updater) =>
|
||
set((state) => ({ runDetails: updater(state.runDetails) })),
|
||
setOpenLogNodeRunIds: (value) =>
|
||
set((state) => ({
|
||
openLogNodeRunIds:
|
||
typeof value === "function"
|
||
? (value as (current: Set<string>) => Set<string>)(
|
||
state.openLogNodeRunIds,
|
||
)
|
||
: value,
|
||
})),
|
||
setLogStates: (updater) =>
|
||
set((state) => ({ logStates: updater(state.logStates) })),
|
||
setArtifactBusyKey: (key) => set({ artifactBusyKey: key }),
|
||
setArtifactErrors: (updater) =>
|
||
set((state) => ({ artifactErrors: updater(state.artifactErrors) })),
|
||
setDetailLoadingId: (id) => set({ detailLoadingId: id }),
|
||
setDetailErrors: (updater) =>
|
||
set((state) => ({ detailErrors: updater(state.detailErrors) })),
|
||
setSelectedNodeId: (id) => set({ selectedNodeId: id }),
|
||
setSelectedEdgeId: (id) => set({ selectedEdgeId: id }),
|
||
setLinkSourceId: (id) => set({ linkSourceId: id }),
|
||
setScheduleKeyword: (k) => set({ scheduleKeyword: k }),
|
||
setArtifactKeyword: (k) => set({ artifactKeyword: k }),
|
||
setCreateDialogOpen: (open) => set({ createDialogOpen: open }),
|
||
setNewScheduleName: (name) => set({ newScheduleName: name }),
|
||
setScheduleForm: (form) => set({ scheduleForm: form }),
|
||
setNodeForm: (form) => set({ nodeForm: form }),
|
||
setCronResult: (result) => set({ cronResult: result }),
|
||
setContextMenu: (menu) => set({ contextMenu: menu }),
|
||
closeContextMenu: () => set({ contextMenu: null }),
|
||
|
||
// ---- data loaders ----
|
||
|
||
loadInitial: async () => {
|
||
const api = requireApi();
|
||
let cancelled = false;
|
||
set({ loading: true });
|
||
try {
|
||
const [scheduleItems, artifactItems] = await Promise.all([
|
||
api.listSchedules(),
|
||
api.listScheduleArtifacts(),
|
||
]);
|
||
if (cancelled) return;
|
||
set({ schedules: scheduleItems, artifacts: artifactItems });
|
||
if (scheduleItems[0]) {
|
||
const detail = await api.getSchedule(scheduleItems[0].schedule_id);
|
||
if (!cancelled) {
|
||
set({ schedule: applyPositionDrafts(detail) });
|
||
}
|
||
}
|
||
setApiOnline(true);
|
||
} catch (error) {
|
||
if (cancelled) return;
|
||
setApiOnline(false);
|
||
notify({
|
||
tone: "error",
|
||
message: error instanceof Error ? error.message : "调度数据加载失败",
|
||
});
|
||
} finally {
|
||
if (!cancelled) set({ loading: false });
|
||
}
|
||
},
|
||
|
||
refreshLists: async (preferredScheduleId) => {
|
||
const api = requireApi();
|
||
const [scheduleItems, artifactItems] = await Promise.all([
|
||
api.listSchedules(),
|
||
api.listScheduleArtifacts(),
|
||
]);
|
||
set({ schedules: scheduleItems, artifacts: artifactItems });
|
||
const targetId =
|
||
preferredScheduleId
|
||
?? get().schedule?.schedule_id
|
||
?? scheduleItems[0]?.schedule_id
|
||
?? null;
|
||
if (!targetId) {
|
||
set({ schedule: null });
|
||
return;
|
||
}
|
||
const detail = await api.getSchedule(targetId);
|
||
set({ schedule: applyPositionDrafts(detail) });
|
||
},
|
||
|
||
refreshRuns: async (scheduleId, showLoading = false) => {
|
||
const api = requireApi();
|
||
if (showLoading) set({ runsLoading: true });
|
||
try {
|
||
const items = await api.listScheduleRuns({ scheduleId, limit: 20 });
|
||
if (get().schedule?.schedule_id === scheduleId) {
|
||
set({ runs: items });
|
||
}
|
||
setApiOnline(true);
|
||
} catch (error) {
|
||
if (showLoading) await handleError(error, "运行记录加载失败");
|
||
} finally {
|
||
if (showLoading) set({ runsLoading: false });
|
||
}
|
||
},
|
||
|
||
chooseSchedule: async (scheduleId) => {
|
||
const api = requireApi();
|
||
const current = get().schedule;
|
||
if (scheduleId === current?.schedule_id || get().busy) return;
|
||
set({ busy: "load-schedule" });
|
||
get().clearAllPositionDrafts();
|
||
set({
|
||
selectedNodeId: null,
|
||
selectedEdgeId: null,
|
||
linkSourceId: null,
|
||
cronResult: null,
|
||
runs: [],
|
||
runsLoading: false,
|
||
});
|
||
try {
|
||
set({ schedule: await api.getSchedule(scheduleId) });
|
||
setApiOnline(true);
|
||
} catch (error) {
|
||
await handleError(error, "调度详情加载失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
// ---- mutations ----
|
||
|
||
openCreateDialog: () => {
|
||
if (get().busy) return;
|
||
set((state) => ({
|
||
contextMenu: null,
|
||
newScheduleName: `新建调度 ${state.schedules.length + 1}`,
|
||
createDialogOpen: true,
|
||
}));
|
||
},
|
||
|
||
addSchedule: async (name) => {
|
||
const api = requireApi();
|
||
const scheduleName = name.trim();
|
||
if (get().busy || !scheduleName) return;
|
||
set({ busy: "create-schedule" });
|
||
try {
|
||
const created = await api.createSchedule({
|
||
schedule_name: scheduleName,
|
||
description: "在画布中拖入稳定版本并配置执行顺序",
|
||
trigger_type: "manual",
|
||
timezone: "Asia/Shanghai",
|
||
enabled: false,
|
||
});
|
||
set((state) => ({
|
||
schedules: [created, ...state.schedules],
|
||
schedule: created,
|
||
createDialogOpen: false,
|
||
newScheduleName: "",
|
||
}));
|
||
get().clearAllPositionDrafts();
|
||
set({ selectedNodeId: null, selectedEdgeId: null });
|
||
notify({ tone: "success", message: "调度方案已创建" });
|
||
} catch (error) {
|
||
await handleError(error, "创建调度失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
removeSchedule: async (target) => {
|
||
const api = requireApi();
|
||
const target_ = target ?? get().schedule;
|
||
if (!target_ || get().busy) return;
|
||
set({ contextMenu: null });
|
||
if (!window.confirm(`确定删除调度"${target_.schedule_name}"吗？`)) return;
|
||
set({ busy: "delete-schedule" });
|
||
try {
|
||
await api.deleteSchedule(target_.schedule_id, target_.workflow_version);
|
||
const remaining = get().schedules.filter(
|
||
(item) => item.schedule_id !== target_.schedule_id,
|
||
);
|
||
set({ schedules: remaining });
|
||
if (get().schedule?.schedule_id === target_.schedule_id) {
|
||
get().clearAllPositionDrafts();
|
||
set({
|
||
schedule: null,
|
||
selectedNodeId: null,
|
||
selectedEdgeId: null,
|
||
});
|
||
if (remaining[0]) {
|
||
const detail = await api.getSchedule(remaining[0].schedule_id);
|
||
set({ schedule: detail });
|
||
}
|
||
}
|
||
notify({ tone: "success", message: "调度方案已删除" });
|
||
} catch (error) {
|
||
await handleError(error, "删除调度失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
renameSchedule: async (target) => {
|
||
const api = requireApi();
|
||
if (get().busy) return;
|
||
set({ contextMenu: null });
|
||
const scheduleName = window
|
||
.prompt("请输入新的调度方案名称", target.schedule_name)
|
||
?.trim();
|
||
if (!scheduleName || scheduleName === target.schedule_name) return;
|
||
set({ busy: "rename-schedule" });
|
||
try {
|
||
const updated = await api.updateSchedule(target.schedule_id, {
|
||
workflow_version: target.workflow_version,
|
||
schedule_name: scheduleName,
|
||
});
|
||
set((state) => ({
|
||
schedules: state.schedules.map((item) =>
|
||
item.schedule_id === updated.schedule_id
|
||
? { ...updated, nodes: [], edges: [] }
|
||
: item
|
||
),
|
||
schedule:
|
||
state.schedule?.schedule_id === updated.schedule_id
|
||
? updated
|
||
: state.schedule,
|
||
}));
|
||
notify({ tone: "success", message: "调度方案已改名" });
|
||
} catch (error) {
|
||
await handleError(error, "调度方案改名失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
removeArtifact: async (artifact) => {
|
||
const api = requireApi();
|
||
if (get().busy) return;
|
||
set({ contextMenu: null });
|
||
if (
|
||
!window.confirm(
|
||
`确定将"${artifact.script_name} ${artifact.version_label}"移出调度列表吗？\n`
|
||
+ "稳定版本本身和历史运行记录不会被删除。",
|
||
)
|
||
) return;
|
||
set({ busy: "delete-artifact" });
|
||
try {
|
||
await api.hideScheduleArtifact(artifact.versions_id);
|
||
set((state) => ({
|
||
artifacts: state.artifacts.filter(
|
||
(item) => item.versions_id !== artifact.versions_id,
|
||
),
|
||
}));
|
||
notify({
|
||
tone: "success",
|
||
message: "已移出调度列表，稳定版本和历史记录保持不变",
|
||
});
|
||
} catch (error) {
|
||
await handleError(error, "移出调度列表失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
saveSchedule: async () => {
|
||
const api = requireApi();
|
||
const state = get();
|
||
const { schedule, scheduleForm } = state;
|
||
if (!schedule || state.busy) return;
|
||
const maxConcurrency = Number(scheduleForm.maxConcurrency);
|
||
if (!scheduleForm.scheduleName.trim()) {
|
||
notify({ tone: "error", message: "调度名称不能为空" });
|
||
return;
|
||
}
|
||
if (!Number.isInteger(maxConcurrency) || maxConcurrency < 1) {
|
||
notify({ tone: "error", message: "最大并发数必须是正整数" });
|
||
return;
|
||
}
|
||
set({ busy: "save-schedule" });
|
||
let updated: Schedule = schedule;
|
||
try {
|
||
for (const [nodeId, position] of positionDrafts.entries()) {
|
||
updated = await api.updateScheduleNode(updated.schedule_id, nodeId, {
|
||
workflow_version: updated.workflow_version,
|
||
position_x: position.position_x,
|
||
position_y: position.position_y,
|
||
});
|
||
}
|
||
updated = await api.updateSchedule(updated.schedule_id, {
|
||
workflow_version: updated.workflow_version,
|
||
schedule_name: scheduleForm.scheduleName.trim(),
|
||
description: scheduleForm.description.trim() || null,
|
||
trigger_type: scheduleForm.triggerType,
|
||
cron_expression:
|
||
scheduleForm.triggerType === "cron"
|
||
? scheduleForm.cronExpression.trim()
|
||
: null,
|
||
timezone: scheduleForm.timezone.trim(),
|
||
enabled: scheduleForm.enabled,
|
||
max_concurrency: maxConcurrency,
|
||
failure_policy: scheduleForm.failurePolicy,
|
||
});
|
||
get().clearAllPositionDrafts();
|
||
set((s) => ({
|
||
schedule: updated,
|
||
schedules: s.schedules.map((item) =>
|
||
item.schedule_id === updated.schedule_id
|
||
? { ...updated, nodes: [], edges: [] }
|
||
: item
|
||
),
|
||
}));
|
||
|
||
notify({ tone: "success", message: "调度配置已保存" });
|
||
} catch (error) {
|
||
const localDraft = applyPositionDrafts(updated);
|
||
set({ schedule: localDraft });
|
||
await handleError(error, "调度配置保存失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
runCronPreview: async () => {
|
||
const api = requireApi();
|
||
const { scheduleForm } = get();
|
||
if (scheduleForm.triggerType !== "cron") return;
|
||
if (get().busy) return;
|
||
set({ busy: "cron-preview" });
|
||
try {
|
||
const result = await api.previewCron({
|
||
cron_expression: scheduleForm.cronExpression.trim(),
|
||
timezone: scheduleForm.timezone.trim(),
|
||
count: 5,
|
||
});
|
||
set({ cronResult: result });
|
||
notify({ tone: "success", message: "Cron 表达式校验通过" });
|
||
} catch (error) {
|
||
set({ cronResult: null });
|
||
await handleError(error, "Cron 预览失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
runNow: async () => {
|
||
const api = requireApi();
|
||
const state = get();
|
||
const { schedule, positionDraftCount } = state;
|
||
if (!schedule || state.busy) return;
|
||
if (positionDraftCount > 0) {
|
||
notify({
|
||
tone: "info",
|
||
message: "还有节点位置未保存，请先点击'保存配置'再运行",
|
||
});
|
||
return;
|
||
}
|
||
if (!schedule.dag_validation.valid || schedule.nodes.length === 0) {
|
||
notify({
|
||
tone: "error",
|
||
message: "当前调度必须包含有效的非空 DAG 才能运行",
|
||
});
|
||
return;
|
||
}
|
||
set({ busy: "run-now" });
|
||
try {
|
||
const created = await api.runScheduleNow(schedule.schedule_id);
|
||
set((s) => ({
|
||
runs: [
|
||
created,
|
||
...s.runs.filter((item) => item.run_id !== created.run_id),
|
||
].slice(0, 20),
|
||
schedule:
|
||
s.schedule
|
||
? { ...s.schedule, last_run_at: created.queued_at }
|
||
: s.schedule,
|
||
schedules: s.schedules.map((item) =>
|
||
item.schedule_id === schedule.schedule_id
|
||
? { ...item, last_run_at: created.queued_at }
|
||
: item
|
||
),
|
||
}));
|
||
notify({
|
||
tone: "success",
|
||
message: `运行 ${created.run_id.slice(-8)} 已进入队列`,
|
||
});
|
||
} catch (error) {
|
||
await handleError(error, "立即运行失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
addArtifactAt: async (artifact, positionX, positionY) => {
|
||
const state = get();
|
||
const { schedule } = state;
|
||
if (!schedule) {
|
||
notify({ tone: "info", message: "请先新建或选择一个调度方案" });
|
||
return;
|
||
}
|
||
const nodeKey = artifactNodeKey(artifact, schedule);
|
||
const updated = await withMutation(
|
||
"add-node",
|
||
() => requireApi().createScheduleNode(schedule.schedule_id, {
|
||
workflow_version: schedule.workflow_version,
|
||
node_key: nodeKey,
|
||
node_name: artifact.script_name,
|
||
versions_id: artifact.versions_id,
|
||
timeout_seconds: 600,
|
||
retry_count: 0,
|
||
retry_interval_sec: 5,
|
||
position_x: Math.max(20, Math.round(positionX)),
|
||
position_y: Math.max(20, Math.round(positionY)),
|
||
arguments_json: {},
|
||
env_refs_json: {},
|
||
python_version: "3.12",
|
||
}),
|
||
`${artifact.script_name} 已加入画布`,
|
||
);
|
||
if (updated) {
|
||
const created = updated.nodes.find((item) => item.node_key === nodeKey);
|
||
set({ selectedNodeId: created?.node_id ?? null });
|
||
}
|
||
},
|
||
|
||
saveNode: async (selectedNode) => {
|
||
const api = requireApi();
|
||
const state = get();
|
||
const { schedule, nodeForm } = state;
|
||
if (!schedule || !selectedNode) return;
|
||
try {
|
||
const timeoutSeconds = Number(nodeForm.timeoutSeconds);
|
||
const retryCount = Number(nodeForm.retryCount);
|
||
const retryIntervalSec = Number(nodeForm.retryIntervalSec);
|
||
if (!nodeForm.nodeName.trim()) throw new Error("节点名称不能为空");
|
||
if (!Number.isInteger(timeoutSeconds) || timeoutSeconds < 1) {
|
||
throw new Error("超时时间必须是正整数");
|
||
}
|
||
if (!Number.isInteger(retryCount) || retryCount < 0) {
|
||
throw new Error("重试次数必须是非负整数");
|
||
}
|
||
if (!Number.isInteger(retryIntervalSec) || retryIntervalSec < 0) {
|
||
throw new Error("重试间隔必须是非负整数");
|
||
}
|
||
if (!PYTHON_VERSION_OPTIONS.includes(nodeForm.pythonVersion)) {
|
||
throw new Error("Python 版本必须是 3.8 / 3.10 / 3.12");
|
||
}
|
||
const argumentsJson = parseObject(nodeForm.argumentsJson, "运行参数");
|
||
const rawEnv = parseObject(nodeForm.envRefsJson, "环境引用");
|
||
const envRefsJson = Object.fromEntries(
|
||
Object.entries(rawEnv).map(([key, value]) => {
|
||
if (typeof value !== "string") {
|
||
throw new Error("环境引用的值必须是字符串");
|
||
}
|
||
return [key, value];
|
||
}),
|
||
);
|
||
await withMutation(
|
||
"save-node",
|
||
() =>
|
||
api.updateScheduleNode(
|
||
schedule.schedule_id,
|
||
selectedNode.node_id,
|
||
{
|
||
workflow_version: schedule.workflow_version,
|
||
node_name: nodeForm.nodeName.trim(),
|
||
timeout_seconds: timeoutSeconds,
|
||
retry_count: retryCount,
|
||
retry_interval_sec: retryIntervalSec,
|
||
arguments_json: argumentsJson,
|
||
env_refs_json: envRefsJson,
|
||
python_version: nodeForm.pythonVersion,
|
||
},
|
||
),
|
||
"节点配置已保存",
|
||
);
|
||
} catch (error) {
|
||
await handleError(error, "节点配置保存失败");
|
||
}
|
||
},
|
||
|
||
removeNode: async (target) => {
|
||
const api = requireApi();
|
||
const state = get();
|
||
const { schedule, selectedNodeId } = state;
|
||
const node = target
|
||
?? schedule?.nodes.find((item) => item.node_id === selectedNodeId)
|
||
?? null;
|
||
if (!schedule || !node || state.busy) return;
|
||
set({ contextMenu: null });
|
||
if (!window.confirm(`确定删除节点"${node.node_name}"吗？`)) return;
|
||
const updated = await withMutation(
|
||
"delete-node",
|
||
() =>
|
||
api.deleteScheduleNode(
|
||
schedule.schedule_id,
|
||
node.node_id,
|
||
schedule.workflow_version,
|
||
),
|
||
"节点已删除",
|
||
);
|
||
if (updated && selectedNodeId === node.node_id) {
|
||
set({ selectedNodeId: null });
|
||
}
|
||
},
|
||
|
||
removeEdge: async (target) => {
|
||
const api = requireApi();
|
||
const state = get();
|
||
const { schedule, selectedEdgeId } = state;
|
||
const edge = target
|
||
?? schedule?.edges.find((item) => item.edge_id === selectedEdgeId)
|
||
?? null;
|
||
if (!schedule || !edge || state.busy) return;
|
||
set({ contextMenu: null });
|
||
const updated = await withMutation(
|
||
"delete-edge",
|
||
() =>
|
||
api.deleteScheduleEdge(
|
||
schedule.schedule_id,
|
||
edge.edge_id,
|
||
schedule.workflow_version,
|
||
),
|
||
"连线已删除",
|
||
);
|
||
if (updated && selectedEdgeId === edge.edge_id) {
|
||
set({ selectedEdgeId: null });
|
||
}
|
||
},
|
||
|
||
checkDag: async () => {
|
||
const api = requireApi();
|
||
const { schedule } = get();
|
||
if (!schedule || get().busy) return;
|
||
set({ busy: "validate" });
|
||
try {
|
||
const result = await api.validateSchedule(schedule.schedule_id);
|
||
set((state) => ({
|
||
schedule: state.schedule
|
||
? { ...state.schedule, dag_validation: result }
|
||
: state.schedule,
|
||
}));
|
||
notify({
|
||
tone: result.valid ? "success" : "error",
|
||
message: result.valid
|
||
? "当前 DAG 通过校验"
|
||
: "当前 DAG 存在校验问题",
|
||
});
|
||
} catch (error) {
|
||
await handleError(error, "DAG 校验失败");
|
||
} finally {
|
||
set({ busy: null });
|
||
}
|
||
},
|
||
|
||
connectTo: async (targetNodeId) => {
|
||
const state = get();
|
||
const { schedule, linkSourceId } = state;
|
||
if (!schedule || !linkSourceId || state.busy) return;
|
||
if (linkSourceId === targetNodeId) {
|
||
set({ linkSourceId: null });
|
||
notify({ tone: "info", message: "已取消连线" });
|
||
return;
|
||
}
|
||
const sourceId = linkSourceId;
|
||
set({ linkSourceId: null });
|
||
await withMutation(
|
||
"create-edge",
|
||
() =>
|
||
requireApi().createScheduleEdge(schedule.schedule_id, {
|
||
workflow_version: schedule.workflow_version,
|
||
source_node_id: sourceId,
|
||
target_node_id: targetNodeId,
|
||
}),
|
||
"节点连线已创建",
|
||
);
|
||
},
|
||
|
||
// ---- runs / logs ----
|
||
|
||
toggleRun: (runId) => {
|
||
const expanded = get().expandedRunId;
|
||
if (expanded === runId) {
|
||
set({ expandedRunId: null });
|
||
} else {
|
||
set({ expandedRunId: runId });
|
||
}
|
||
},
|
||
setExpandedRunId: (id) =>
|
||
set((state) => ({
|
||
expandedRunId:
|
||
typeof id === "function"
|
||
? (id as (current: string | null) => string | null)(
|
||
state.expandedRunId,
|
||
)
|
||
: id,
|
||
})),
|
||
|
||
toggleLog: async (runId, nodeRunId) => {
|
||
const api = requireApi();
|
||
const state = get();
|
||
const open = state.openLogNodeRunIds;
|
||
if (open.has(nodeRunId)) {
|
||
const next = new Set(open);
|
||
next.delete(nodeRunId);
|
||
set({ openLogNodeRunIds: next });
|
||
return;
|
||
}
|
||
const next = new Set(open);
|
||
next.add(nodeRunId);
|
||
set({ openLogNodeRunIds: next });
|
||
if (state.logStates[nodeRunId]?.content) return;
|
||
set((s) => ({
|
||
logStates: {
|
||
...s.logStates,
|
||
[nodeRunId]: {
|
||
loading: true,
|
||
content: null,
|
||
fileName: null,
|
||
error: null,
|
||
},
|
||
},
|
||
}));
|
||
try {
|
||
const artifacts = await api.getScheduleNodeRunArtifacts(runId, nodeRunId);
|
||
if (!artifacts.log) throw new Error("本次节点执行没有可用日志");
|
||
const response = await fetch(artifacts.log.url, {
|
||
credentials: "same-origin",
|
||
cache: "no-store",
|
||
});
|
||
if (!response.ok) {
|
||
throw new Error(`日志读取失败（HTTP ${response.status}）`);
|
||
}
|
||
const content = await response.text();
|
||
set((s) => ({
|
||
logStates: {
|
||
...s.logStates,
|
||
[nodeRunId]: {
|
||
loading: false,
|
||
content,
|
||
fileName: artifacts.log?.file_name ?? null,
|
||
error: null,
|
||
},
|
||
},
|
||
}));
|
||
} catch (error) {
|
||
set((s) => ({
|
||
logStates: {
|
||
...s.logStates,
|
||
[nodeRunId]: {
|
||
loading: false,
|
||
content: null,
|
||
fileName: null,
|
||
error: error instanceof Error ? error.message : "日志读取失败",
|
||
},
|
||
},
|
||
}));
|
||
}
|
||
},
|
||
|
||
downloadResult: async (runId, nodeRunId) => {
|
||
const api = requireApi();
|
||
const busyKey = `${nodeRunId}:result`;
|
||
set({
|
||
artifactBusyKey: busyKey,
|
||
artifactErrors: (() => {
|
||
const next = { ...get().artifactErrors };
|
||
delete next[nodeRunId];
|
||
return next;
|
||
})(),
|
||
});
|
||
try {
|
||
const artifacts = await api.getScheduleNodeRunArtifacts(runId, nodeRunId);
|
||
if (!artifacts.result) throw new Error("本次节点执行没有可下载结果");
|
||
const link = document.createElement("a");
|
||
link.href = artifacts.result.url;
|
||
link.download = artifacts.result.file_name;
|
||
document.body.appendChild(link);
|
||
link.click();
|
||
link.remove();
|
||
} catch (error) {
|
||
set((state) => ({
|
||
artifactErrors: {
|
||
...state.artifactErrors,
|
||
[nodeRunId]: error instanceof Error ? error.message : "结果下载失败",
|
||
},
|
||
}));
|
||
} finally {
|
||
set({ artifactBusyKey: null });
|
||
}
|
||
},
|
||
|
||
downloadLog: async (runId, nodeRunId) => {
|
||
const api = requireApi();
|
||
const busyKey = `${nodeRunId}:log-download`;
|
||
set({
|
||
artifactBusyKey: busyKey,
|
||
artifactErrors: (() => {
|
||
const next = { ...get().artifactErrors };
|
||
delete next[nodeRunId];
|
||
return next;
|
||
})(),
|
||
});
|
||
try {
|
||
const artifacts = await api.getScheduleNodeRunArtifacts(runId, nodeRunId);
|
||
if (!artifacts.log) throw new Error("本次节点执行没有可下载日志");
|
||
const response = await fetch(artifacts.log.url, {
|
||
credentials: "same-origin",
|
||
cache: "no-store",
|
||
});
|
||
if (!response.ok) {
|
||
throw new Error(`日志下载失败（HTTP ${response.status}）`);
|
||
}
|
||
const objectUrl = URL.createObjectURL(await response.blob());
|
||
const link = document.createElement("a");
|
||
link.href = objectUrl;
|
||
link.download = artifacts.log.file_name;
|
||
document.body.appendChild(link);
|
||
link.click();
|
||
link.remove();
|
||
URL.revokeObjectURL(objectUrl);
|
||
} catch (error) {
|
||
set((state) => ({
|
||
artifactErrors: {
|
||
...state.artifactErrors,
|
||
[nodeRunId]:
|
||
error instanceof Error ? error.message : "日志下载失败",
|
||
},
|
||
}));
|
||
} finally {
|
||
set({ artifactBusyKey: null });
|
||
}
|
||
},
|
||
|
||
// ---- drag helpers ----
|
||
|
||
setPositionDraft: (nodeId, draft) => {
|
||
positionDrafts.set(nodeId, draft);
|
||
set({ positionDraftCount: positionDrafts.size });
|
||
},
|
||
setPositionDraftCount: (count) => set({ positionDraftCount: count }),
|
||
getPositionDrafts: () => positionDrafts,
|
||
clearAllPositionDrafts: () => {
|
||
clearPositionDrafts();
|
||
set({ positionDraftCount: 0 });
|
||
},
|
||
prunePositionDrafts: (validNodeIds) => {
|
||
for (const nodeId of [...positionDrafts.keys()]) {
|
||
if (!validNodeIds.has(nodeId)) positionDrafts.delete(nodeId);
|
||
}
|
||
set({ positionDraftCount: positionDrafts.size });
|
||
},
|
||
};
|
||
});
|
||
|
||
// 顶层 pure helper (scheduleNodeKey needs schedule, so it's defined outside the store closure)
|
||
function artifactNodeKey(artifact: ScheduleArtifact, schedule: Schedule): string {
|
||
const ascii = artifact.script_name
|
||
.toLowerCase()
|
||
.replace(/[^a-z0-9_]+/g, "_")
|
||
.replace(/^_+|_+$/g, "")
|
||
.slice(0, 24);
|
||
const base = ascii || `node_${artifact.versions_id.slice(-6).toLowerCase()}`;
|
||
const existing = new Set(schedule.nodes.map((item) => item.node_key));
|
||
if (!existing.has(base)) return base;
|
||
for (let i = 2; i < 1000; i++) {
|
||
const candidate = `${base}_${i}`;
|
||
if (!existing.has(candidate)) return candidate;
|
||
}
|
||
return `${base}_${Date.now().toString(36)}`;
|
||
}
|