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

import { useCallback, useSyncExternalStore } from "react"

/**
 * 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
}
/** 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 health = new Map<string, NodeHealth>()
let engineEvents: EngineEvent[] = []
// How many times a node has emitted. The number itself means nothing; a change
// is what restarts the pulse.
const emits = 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}`)
  },
  setStatuses(
    entries: { id: string; status: string; error?: string | null }[],
  ) {
    for (const entry of entries) {
      statuses.set(entry.id, {
        status: entry.status as LiveStatus["status"],
        error: entry.error,
      })
      notify(`status:${entry.id}`)
    }
  },
  getStatus(nodeId: string) {
    return statuses.get(nodeId)
  },
  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}`)
  },
  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 emits.keys()) notify(`emit:${key}`)
    emits.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),
  )
}

/** 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 = ""

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(
          joined.split(SEP).map((id) => `emit:${id}`),
          listener,
        ),
      [joined],
    ),
    () =>
      joined.split(SEP).reduce((total, id) => total + (emits.get(id) ?? 0), 0),
  )
}

/** Whether any of these nodes is currently failing. */
export function useGroupError(ids: string[]): boolean {
  const joined = ids.join(SEP)
  return useSyncExternalStore(
    useCallback(
      (listener: Listener) =>
        subscribeAll(
          joined.split(SEP).map((id) => `status:${id}`),
          listener,
        ),
      [joined],
    ),
    () => joined.split(SEP).some((id) => statuses.get(id)?.status === "error"),
  )
}

/** 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(
          joined.split(SEP).map((name) => `value:${name}`),
          listener,
        ),
      [joined],
    ),
    () =>
      joined
        .split(SEP)
        .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,
  )
}