stroblme
0be8bf5713
Suggest a parameter's value from the nodes of the same type in every flow, the way message names are already offered; secrets stay out of it. Explain the cron fields, and derive the expression an interval asks for. Keep an untouched function scaffold in step with the node's ports.
272 lines
9.4 KiB
TypeScript
272 lines
9.4 KiB
TypeScript
import {
|
|
type UseMutationResult,
|
|
useMutation,
|
|
useQueries,
|
|
useQuery,
|
|
useQueryClient,
|
|
} from "@tanstack/react-query"
|
|
import { useCallback, useEffect, useRef, useState } from "react"
|
|
|
|
import {
|
|
ApiError,
|
|
type FlowDef_Input,
|
|
FlowsService,
|
|
SecretsService,
|
|
} from "@/client"
|
|
|
|
export const flowKeys = {
|
|
all: ["flows"] as const,
|
|
detail: (name: string) => ["flows", name] as const,
|
|
source: (name: string, nodeId: string) =>
|
|
["flows", name, "source", nodeId] as const,
|
|
history: (name: string, message: string) =>
|
|
["flows", name, "history", message] as const,
|
|
nodeTypes: ["flows", "node-types"] as const,
|
|
library: ["flows", "library"] as const,
|
|
}
|
|
|
|
export const flowsQueryOptions = () => ({
|
|
queryKey: flowKeys.all,
|
|
queryFn: () => FlowsService.readFlows(),
|
|
})
|
|
|
|
export const flowQueryOptions = (name: string) => ({
|
|
queryKey: flowKeys.detail(name),
|
|
queryFn: () => FlowsService.readFlow({ name }),
|
|
})
|
|
|
|
/** Names of the stored secrets, for pointing a credential parameter at one. */
|
|
export const secretsQueryOptions = () => ({
|
|
queryKey: ["secrets"] as const,
|
|
queryFn: () => SecretsService.readSecrets(),
|
|
})
|
|
|
|
/** Node sources shared across flows, with the nodes using each. */
|
|
export const libraryQueryOptions = () => ({
|
|
queryKey: flowKeys.library,
|
|
queryFn: () => FlowsService.readLibrary(),
|
|
})
|
|
|
|
export const nodeTypesQueryOptions = () => ({
|
|
queryKey: flowKeys.nodeTypes,
|
|
queryFn: () => FlowsService.readNodeTypes(),
|
|
staleTime: Number.POSITIVE_INFINITY,
|
|
})
|
|
|
|
export const nodeSourceQueryOptions = (name: string, nodeId: string) => ({
|
|
queryKey: flowKeys.source(name, nodeId),
|
|
queryFn: () => FlowsService.readNodeSource({ name, nodeId }),
|
|
})
|
|
|
|
/** The recent values of one message, for the panel's sparkline. */
|
|
export const messageHistoryQueryOptions = (name: string, message: string) => ({
|
|
queryKey: flowKeys.history(name, message),
|
|
queryFn: () => FlowsService.readMessageHistory({ name, message }),
|
|
})
|
|
|
|
/** Other flows change rarely; suggestions do not need them fresh to the second. */
|
|
const SUGGEST_STALE = 5 * 60 * 1000
|
|
|
|
/**
|
|
* Values already in use by nodes of the same type, keyed by parameter.
|
|
*
|
|
* The second MQTT node points at the same broker as the first one, so the
|
|
* settings are worth offering rather than making someone type them again.
|
|
*
|
|
* Only plain values are collected: a secret is stored as a reference object
|
|
* and stays out of this, which is the point of the secret picker.
|
|
*/
|
|
export function useParamSuggestions(
|
|
type: string | undefined,
|
|
): Record<string, string[]> {
|
|
const { data: flows } = useQuery({
|
|
...flowsQueryOptions(),
|
|
staleTime: SUGGEST_STALE,
|
|
})
|
|
// ponytail: reads every flow to collect them, sharing the editor's own cache
|
|
// entries; an aggregate endpoint if a big installation makes that hurt.
|
|
const details = useQueries({
|
|
queries: (type ? (flows?.data ?? []) : []).map((flow) => ({
|
|
...flowQueryOptions(flow.name),
|
|
staleTime: SUGGEST_STALE,
|
|
})),
|
|
})
|
|
|
|
const seen: Record<string, Set<string>> = {}
|
|
for (const { data } of details) {
|
|
for (const node of data?.definition.nodes ?? []) {
|
|
if (node.type !== type) continue
|
|
for (const [key, value] of Object.entries(node.params ?? {})) {
|
|
if (value === null || value === "" || typeof value === "object")
|
|
continue
|
|
if (!seen[key]) seen[key] = new Set()
|
|
seen[key].add(String(value))
|
|
}
|
|
}
|
|
}
|
|
return Object.fromEntries(
|
|
Object.entries(seen).map(([key, values]) => [key, [...values].sort()]),
|
|
)
|
|
}
|
|
|
|
const AUTOSAVE_DELAY = 800
|
|
/** How long to wait for a save in flight before sending the next one. */
|
|
const RETRY_DELAY = 100
|
|
|
|
/** Publish the unpublished changes, which is what puts them on the engine. */
|
|
export function usePublish(name: string) {
|
|
const queryClient = useQueryClient()
|
|
return useMutation({
|
|
mutationFn: (version: number) =>
|
|
FlowsService.publishFlow({ name, requestBody: { version } }),
|
|
onSuccess: (detail) => {
|
|
queryClient.setQueryData(flowKeys.detail(name), detail)
|
|
queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
|
|
},
|
|
})
|
|
}
|
|
|
|
/** Throw the unpublished changes away and go back to what is running. */
|
|
export function useDiscardDraft(name: string) {
|
|
const queryClient = useQueryClient()
|
|
return useMutation({
|
|
mutationFn: () => FlowsService.discardDraft({ name }),
|
|
onSuccess: (detail) => {
|
|
queryClient.setQueryData(flowKeys.detail(name), detail)
|
|
queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
|
|
},
|
|
})
|
|
}
|
|
|
|
/**
|
|
* Saves the flow a moment after the last edit, and immediately when the editor
|
|
* needs the server to be current (closing a panel, switching flow, running).
|
|
*
|
|
* Saving writes a draft — the engine keeps running the published version until
|
|
* someone publishes. Identical documents are skipped server-side, so a quiet
|
|
* canvas writes nothing.
|
|
*
|
|
* Every save carries the version it is based on. If another client saved in
|
|
* between, the server refuses rather than discarding their work, and `conflict`
|
|
* turns true for the editor to ask which version wins.
|
|
*/
|
|
export function useAutosave(name: string): {
|
|
save: (definition: FlowDef_Input) => void
|
|
/** Send what is queued and resolve once the server has it. */
|
|
flush: () => Promise<void>
|
|
conflict: boolean
|
|
resolveConflict: (mode: "theirs" | "mine") => Promise<void>
|
|
mutation: UseMutationResult<unknown, unknown, FlowDef_Input, unknown>
|
|
} {
|
|
const queryClient = useQueryClient()
|
|
const pending = useRef<FlowDef_Input | null>(null)
|
|
const timer = useRef<ReturnType<typeof setTimeout> | null>(null)
|
|
// The freshest version this client has seen. Local edits keep the document
|
|
// they were made against, so the version has to be stamped on at send time.
|
|
const version = useRef<number | null>(null)
|
|
const sent = useRef<FlowDef_Input | null>(null)
|
|
const inFlight = useRef(false)
|
|
const [conflict, setConflict] = useState(false)
|
|
|
|
const mutation = useMutation({
|
|
mutationFn: (definition: FlowDef_Input) =>
|
|
FlowsService.saveFlow({
|
|
name,
|
|
// The ref holds the freshest version this client has seen; before the
|
|
// first save that is the one the document was loaded with. Sending
|
|
// `undefined` would drop the field and let the server read the default,
|
|
// which conflicts with every flow saved more than once.
|
|
requestBody: {
|
|
...definition,
|
|
version: version.current ?? definition.version ?? 1,
|
|
},
|
|
}),
|
|
onMutate: (definition) => {
|
|
inFlight.current = true
|
|
sent.current = definition
|
|
},
|
|
onSettled: () => {
|
|
inFlight.current = false
|
|
},
|
|
onSuccess: (detail) => {
|
|
version.current = detail.definition.version ?? null
|
|
// Write the server's answer straight into the cache: invalidating would
|
|
// pull the document back out from under edits still in flight.
|
|
queryClient.setQueryData(flowKeys.detail(name), detail)
|
|
queryClient.invalidateQueries({ queryKey: flowKeys.all, exact: true })
|
|
},
|
|
onError: (error) => {
|
|
if (error instanceof ApiError && error.status === 409) setConflict(true)
|
|
},
|
|
})
|
|
|
|
// react-query hands back a new mutation object on every render, so flushing
|
|
// has to hang off `mutate`, which is stable. Depending on the whole mutation
|
|
// re-ran the effect below on every render, and its cleanup cancelled the
|
|
// pending save before it ever fired.
|
|
const { mutate, mutateAsync } = mutation
|
|
const flush = useCallback(async (): Promise<void> => {
|
|
if (timer.current) {
|
|
clearTimeout(timer.current)
|
|
timer.current = null
|
|
}
|
|
const definition = pending.current
|
|
if (!definition) return
|
|
// Two saves in flight at once would race for the same version, and the
|
|
// loser would look like someone else's edit.
|
|
if (inFlight.current) {
|
|
await new Promise((resolve) => setTimeout(resolve, RETRY_DELAY))
|
|
return flush()
|
|
}
|
|
pending.current = null
|
|
// A rejection here is a conflict, which the dialog handles; callers waiting
|
|
// on the flush only need to know the attempt is over.
|
|
await mutateAsync(definition).catch(() => undefined)
|
|
}, [mutateAsync])
|
|
|
|
const resolveConflict = useCallback(
|
|
async (mode: "theirs" | "mine") => {
|
|
setConflict(false)
|
|
if (mode === "theirs") {
|
|
pending.current = null
|
|
sent.current = null
|
|
version.current = null
|
|
await queryClient.invalidateQueries({
|
|
queryKey: flowKeys.detail(name),
|
|
})
|
|
return
|
|
}
|
|
const current = await FlowsService.readFlow({ name })
|
|
version.current = current.definition.version ?? null
|
|
const definition = pending.current ?? sent.current
|
|
pending.current = null
|
|
if (definition) mutate(definition)
|
|
},
|
|
[mutate, name, queryClient],
|
|
)
|
|
|
|
const save = useCallback(
|
|
(definition: FlowDef_Input) => {
|
|
pending.current = definition
|
|
if (timer.current) clearTimeout(timer.current)
|
|
timer.current = setTimeout(() => void flush(), AUTOSAVE_DELAY)
|
|
},
|
|
[flush],
|
|
)
|
|
|
|
// Leaving the tab is the last chance to persist what is still queued.
|
|
useEffect(() => {
|
|
const onHidden = () => {
|
|
if (document.visibilityState === "hidden") void flush()
|
|
}
|
|
document.addEventListener("visibilitychange", onHidden)
|
|
return () => {
|
|
document.removeEventListener("visibilitychange", onHidden)
|
|
// Unmounting is a flow switch, not a reason to drop a queued edit.
|
|
void flush()
|
|
}
|
|
}, [flush])
|
|
|
|
return { save, flush, conflict, resolveConflict, mutation }
|
|
}
|