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

import { useCallback, useSyncExternalStore } from "react"

import { OpenAPI } from "@/client"
import { request } from "@/client/core/request"

/**
 * Live engine state, deliberately outside React Query.
 *
 * Values can arrive many times a second. Keeping them here, with one subscriber
 * set per key, means a changing value re-renders the chip showing it and
 * nothing else.
 */

/** Who caused a value. The canvas needs it to pulse the right edge. */
export type ValueSource = {
  kind: "node" | "dashboard" | "flow" | "agent" | "api"
  id: string
  label: string
  detail?: string
}
export type LiveValue = {
  value: unknown
  ts: number | null
  source?: ValueSource
}
export type LiveStatus = {
  status: "active" | "error" | "running" | "success"
  error?: string | null
}
/** How a node's connection is doing, which is not how its last run went. */
export type NodeHealth = {
  health: "ok" | "down" | "unknown"
  detail?: string | null
}
/**
 * A node's last failure, kept after it has run again.
 *
 * `LiveStatus` answers "how did the last run go", which is what the red dot
 * has to keep telling the truth about. This answers the other question — did
 * this node fail at all since anyone looked — and the engine answers it too,
 * so a reload and a second browser see the same thing. Only acknowledging one
 * clears it.
 */
export type NodeFailure = { error: string; ts: number }
/** Something the engine reported about itself, for the health page. */
export type EngineEvent = {
  type: string
  flow?: string
  node?: string
  detail?: string
  ts: number
}
/** One node execution's output, as the log panel shows it. */
export type LogLine = {
  flow: string
  node: string
  text: string
  level: "info" | "error"
  truncated?: boolean
  ts: number
}

type Listener = () => void

/** Enough to see what a flow has been doing, not a log store. */
const LOG_LIMIT = 500
/** The health page reads these to know when to refetch; it is not a history. */
const ENGINE_EVENT_LIMIT = 100

const values = new Map<string, LiveValue>()
const statuses = new Map<string, LiveStatus>()
const failures = new Map<string, NodeFailure>()
const health = new Map<string, NodeHealth>()
let engineEvents: EngineEvent[] = []
// How many times this page has seen a node emit. The number itself means
// nothing; a change is what restarts the pulse.
const emits = new Map<string, number>()
// What the engine had already counted when we connected, from the snapshot.
// Kept apart from the live count on purpose: it says the node has been busy,
// which is worth drawing, but it happened before anyone was watching, and a
// pulse claims something is happening now.
const priorEmits = new Map<string, number>()
let logLines: LogLine[] = []
const paused = new Set<string>()
const listeners = new Map<string, Set<Listener>>()

let connected = false
const connectionListeners = new Set<Listener>()

function notify(key: string) {
  for (const listener of listeners.get(key) ?? []) listener()
}

function subscribeKey(key: string, listener: Listener) {
  let set = listeners.get(key)
  if (!set) {
    set = new Set()
    listeners.set(key, set)
  }
  set.add(listener)
  return () => {
    set.delete(listener)
    if (set.size === 0) listeners.delete(key)
  }
}

export const liveStore = {
  setValue(name: string, value: LiveValue) {
    values.set(name, value)
    notify(`value:${name}`)
  },
  setValues(entries: Record<string, LiveValue>) {
    for (const [name, value] of Object.entries(entries)) {
      values.set(name, value)
      notify(`value:${name}`)
    }
  },
  getValue(name: string) {
    return values.get(name)
  },
  setStatus(nodeId: string, status: LiveStatus) {
    statuses.set(nodeId, status)
    notify(`status:${nodeId}`)
    // Recorded here rather than waited for: the engine keeps the same record,
    // but the canvas has to mark the failure as it happens. A node only ever
    // reaches this store as "error" by having raised.
    if (status.status === "error") {
      failures.set(nodeId, {
        error: status.error || "The node raised while running.",
        ts: Date.now() / 1000,
      })
      notify(`failure:${nodeId}`)
    }
  },
  setStatuses(
    entries: {
      id: string
      status: string
      error?: string | null
      /** The engine's own record of the last failure — see `NodeFailure`. */
      last_error?: string | null
      last_error_ts?: number | null
    }[],
  ) {
    for (const entry of entries) {
      statuses.set(entry.id, {
        status: entry.status as LiveStatus["status"],
        error: entry.error,
      })
      notify(`status:${entry.id}`)
      // The engine's record wins, in both directions: it is what a reload
      // reads, and an empty one means someone has acknowledged the failure.
      if (entry.last_error) {
        failures.set(entry.id, {
          error: entry.last_error,
          ts: entry.last_error_ts ?? Date.now() / 1000,
        })
        notify(`failure:${entry.id}`)
      } else if (failures.delete(entry.id)) {
        notify(`failure:${entry.id}`)
      }
    }
  },
  getStatus(nodeId: string) {
    return statuses.get(nodeId)
  },
  getFailure(nodeId: string) {
    return failures.get(nodeId)
  },
  /**
   * Dismiss a node's failure, here and on the engine.
   *
   * The only thing that clears one: a good run afterwards deliberately does
   * not, which is what makes a failure between two glances at the canvas
   * findable at all.
   */
  acknowledgeFailure(nodeId: string) {
    if (!failures.delete(nodeId)) return
    notify(`failure:${nodeId}`)
    const [flow, node] = nodeId.split(".")
    if (!flow || !node) return
    // Hand-written rather than generated: nothing hangs off the response, and
    // a call that does not arrive only means the marker is back after a
    // reload — which is the safer way round for a failure.
    request(OpenAPI, {
      method: "POST",
      url: "/api/v1/flows/{name}/nodes/{node_id}/acknowledge",
      path: { name: flow, node_id: node },
    }).catch(() => {})
  },
  setHealth(nodeId: string, entry: NodeHealth) {
    health.set(nodeId, entry)
    notify(`health:${nodeId}`)
  },
  getHealth(nodeId: string) {
    return health.get(nodeId)
  },
  recordEngineEvent(event: EngineEvent) {
    // A new array each time, so the hook's snapshot comparison sees the change.
    engineEvents = [...engineEvents, event].slice(-ENGINE_EVENT_LIMIT)
    notify("engine")
  },
  recordEmit(nodeId: string) {
    emits.set(nodeId, (emits.get(nodeId) ?? 0) + 1)
    notify(`emit:${nodeId}`)
  },
  /**
   * The counts a snapshot carries, so a page that connects into an engine
   * already at work draws what has been busy rather than a graph of idle
   * neurons.
   *
   * Absent keys are left alone rather than cleared: a restarted engine counts
   * from zero again, which is no reason to forget what this page saw.
   */
  setEmits(entries: Record<string, number>) {
    for (const [nodeId, count] of Object.entries(entries)) {
      if (priorEmits.get(nodeId) === count) continue
      priorEmits.set(nodeId, count)
      notify(`emit:${nodeId}`)
    }
  },
  appendLog(line: LogLine) {
    // A new array each time, so the hook's snapshot comparison sees the change.
    logLines = [...logLines, line].slice(-LOG_LIMIT)
    notify("logs")
  },
  setLogs(lines: LogLine[]) {
    logLines = lines.slice(-LOG_LIMIT)
    notify("logs")
  },
  clearLogs() {
    logLines = []
    notify("logs")
  },
  setPaused(flow: string, isPaused: boolean) {
    if (isPaused) paused.add(flow)
    else paused.delete(flow)
    notify(`paused:${flow}`)
  },
  setPausedFlows(flows: string[]) {
    const next = new Set(flows)
    for (const flow of new Set([...paused, ...next])) {
      if (paused.has(flow) === next.has(flow)) continue
      if (next.has(flow)) paused.add(flow)
      else paused.delete(flow)
      notify(`paused:${flow}`)
    }
  },
  setConnected(next: boolean) {
    if (connected === next) return
    connected = next
    for (const listener of connectionListeners) listener()
  },
  isConnected() {
    return connected
  },
  reset() {
    for (const key of values.keys()) notify(`value:${key}`)
    values.clear()
    for (const key of statuses.keys()) notify(`status:${key}`)
    statuses.clear()
    for (const key of failures.keys()) notify(`failure:${key}`)
    failures.clear()
    for (const key of new Set([...emits.keys(), ...priorEmits.keys()]))
      notify(`emit:${key}`)
    emits.clear()
    priorEmits.clear()
    for (const key of health.keys()) notify(`health:${key}`)
    health.clear()
    engineEvents = []
    notify("engine")
    logLines = []
    notify("logs")
    for (const flow of paused) notify(`paused:${flow}`)
    paused.clear()
  },
}

export function useLiveValue(name: string | undefined): LiveValue | undefined {
  return useSyncExternalStore(
    (listener) => (name ? subscribeKey(`value:${name}`, listener) : () => {}),
    () => (name ? values.get(name) : undefined),
  )
}

export function useNodeStatus(nodeId: string): LiveStatus | undefined {
  return useSyncExternalStore(
    (listener) => subscribeKey(`status:${nodeId}`, listener),
    () => statuses.get(nodeId),
  )
}

/** The node's last failure, until someone acknowledges it. */
export function useNodeFailure(nodeId: string): NodeFailure | undefined {
  return useSyncExternalStore(
    (listener) => subscribeKey(`failure:${nodeId}`, listener),
    () => failures.get(nodeId),
  )
}

/** How the node's connection is doing, once it has said anything about it. */
export function useNodeHealth(nodeId: string): NodeHealth | undefined {
  return useSyncExternalStore(
    (listener) => subscribeKey(`health:${nodeId}`, listener),
    () => health.get(nodeId),
  )
}

/** The last hundred things the engine said about itself, oldest first. */
export function useEngineEvents(): EngineEvent[] {
  return useSyncExternalStore(
    (listener) => subscribeKey("engine", listener),
    () => engineEvents,
  )
}

/** Increments each time the node publishes something. */
export function useNodeEmits(nodeId: string): number {
  return useSyncExternalStore(
    (listener) => subscribeKey(`emit:${nodeId}`, listener),
    () => emits.get(nodeId) ?? 0,
  )
}

/** Every captured line, newest last. Filtered by the panel that shows it. */
export function useLiveLogs(): LogLine[] {
  return useSyncExternalStore(
    (listener) => subscribeKey("logs", listener),
    () => logLines,
  )
}

export function useFlowPaused(flow: string): boolean {
  return useSyncExternalStore(
    (listener) => subscribeKey(`paused:${flow}`, listener),
    () => paused.has(flow),
  )
}

/*
 * Group hooks, for the brain graph: one neuron stands for several nodes, and
 * one edge for several messages.
 *
 * Each returns a primitive. `useSyncExternalStore` compares snapshots by
 * identity, so a fresh object or array every render would re-render forever.
 * The names arrive as an array, which is also new every render, so they are
 * joined into one string before anything hangs off them.
 */

const SEP = "\u0000"

/** Back to the array. `"".split(SEP)` is `[""]`, which is one empty name. */
const parts = (joined: string) => (joined ? joined.split(SEP) : [])

function subscribeAll(keys: string[], listener: Listener) {
  const unsubscribes = keys.map((key) => subscribeKey(key, listener))
  return () => {
    for (const unsubscribe of unsubscribes) unsubscribe()
  }
}

/** How many times any of these nodes has published. A change is a pulse. */
export function useGroupEmits(ids: string[]): number {
  const joined = ids.join(SEP)
  return useSyncExternalStore(
    useCallback(
      (listener: Listener) =>
        subscribeAll(
          parts(joined).map((id) => `emit:${id}`),
          listener,
        ),
      [joined],
    ),
    () => parts(joined).reduce((total, id) => total + (emits.get(id) ?? 0), 0),
  )
}

/**
 * Whether any of these nodes has ever published, this page or before it.
 *
 * Separate from the count above because it answers a different question: the
 * count is what pulses, and this is what tells a neuron nobody has ever seen
 * run from one that is merely quiet just now.
 */
export function useGroupActive(ids: string[]): boolean {
  const joined = ids.join(SEP)
  return useSyncExternalStore(
    useCallback(
      (listener: Listener) =>
        subscribeAll(
          parts(joined).map((id) => `emit:${id}`),
          listener,
        ),
      [joined],
    ),
    () =>
      parts(joined).some(
        (id) => (emits.get(id) ?? 0) + (priorEmits.get(id) ?? 0) > 0,
      ),
  )
}

/**
 * What any of these nodes last failed with, or `""` for none.
 *
 * The failure rather than the live status: a neuron stands for nodes across
 * several flows, and Home would otherwise disagree with the canvas the moment
 * one of them ran again. The reason itself, because a string is a snapshot
 * `useSyncExternalStore` can compare and the neuron shows it in its tooltip.
 */
export function useGroupFailure(ids: string[]): string {
  const joined = ids.join(SEP)
  return useSyncExternalStore(
    useCallback(
      (listener: Listener) =>
        subscribeAll(
          parts(joined).map((id) => `failure:${id}`),
          listener,
        ),
      [joined],
    ),
    () =>
      parts(joined)
        .map((id) => failures.get(id)?.error ?? "")
        .find(Boolean) ?? "",
  )
}

/** When any of these messages last carried a value. */
export function useLatestTs(names: string[]): number {
  const joined = names.join(SEP)
  return useSyncExternalStore(
    useCallback(
      (listener: Listener) =>
        subscribeAll(
          parts(joined).map((name) => `value:${name}`),
          listener,
        ),
      [joined],
    ),
    () =>
      parts(joined).reduce(
        (latest, name) => Math.max(latest, values.get(name)?.ts ?? 0),
        0,
      ),
  )
}

export function useLiveConnection(): boolean {
  return useSyncExternalStore(
    (listener) => {
      connectionListeners.add(listener)
      return () => connectionListeners.delete(listener)
    },
    () => connected,
  )
}