import { type UseMutationResult, useMutation, useQueries, useQuery, useQueryClient, } from "@tanstack/react-query" import { useCallback, useEffect, useRef, useState } from "react" import { ApiError, FlavorsService, 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, // Under `all` on purpose: the socket invalidates that prefix on every // rebuild, so the brain graph re-layouts when someone publishes. graph: ["flows", "graph"] 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(), }) /** The named sizes a node can ask for, offered in the node panel. */ export const flavorsQueryOptions = () => ({ queryKey: ["flavors"] as const, queryFn: () => FlavorsService.readFlavors(), }) /** Every flow at once, merged on what its nodes talk to. */ export const graphQueryOptions = () => ({ queryKey: flowKeys.graph, queryFn: () => FlowsService.readGraph(), }) /** 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 { 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 instance makes that hurt. const details = useQueries({ queries: (type ? (flows?.data ?? []) : []).map((flow) => ({ ...flowQueryOptions(flow.name), staleTime: SUGGEST_STALE, })), }) const seen: Record> = {} 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 conflict: boolean resolveConflict: (mode: "theirs" | "mine") => Promise /** Take a version the server moved the draft to by itself. */ adopt: (next: number) => void mutation: UseMutationResult } { const queryClient = useQueryClient() const pending = useRef(null) const timer = useRef | 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(null) const sent = useRef(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 => { 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]) const adopt = useCallback((next: number) => { version.current = next }, []) return { save, flush, conflict, resolveConflict, adopt, mutation } }