app/frontend/src/components/Flow/queries.ts

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,
  // 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(),
})

/** 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<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 }
}