import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react' import { GitBranch, Lock, LockOpen, Play, RefreshCw, ShieldAlert, ShieldCheck, Loader2, Bot } from 'lucide-react' import type { DataflowEdge, DataflowGraph, DataflowNode } from '../../types' import { fetchDataflow, runDataflowMovement, toggleEtlAgent, fetchAgentOpsStatus, setPiiMask } from '../../lib/api' import { Badge } from '../ui/Badge' import { cn } from '../../lib/utils' /* ── styling maps ───────────────────────────────────────────────── */ const NODE_KIND: Record = { generator: { ring: 'border-amber-400/60', chip: 'bg-amber-500/15 text-amber-300 border-amber-400/40', dot: '#f59e0b' }, hadoop: { ring: 'border-orange-400/60', chip: 'bg-orange-500/15 text-orange-300 border-orange-400/40', dot: '#fb923c' }, source: { ring: 'border-emerald-400/60', chip: 'bg-emerald-500/15 text-emerald-300 border-emerald-400/40', dot: '#34d399' }, stream: { ring: 'border-cyan-400/60', chip: 'bg-cyan-500/15 text-cyan-300 border-cyan-400/40', dot: '#22d3ee' }, sink: { ring: 'border-sky-400/60', chip: 'bg-sky-500/15 text-sky-300 border-sky-400/40', dot: '#38bdf8' }, lakehouse: { ring: 'border-blue-400/60', chip: 'bg-blue-500/15 text-blue-300 border-blue-400/40', dot: '#60a5fa' }, engine: { ring: 'border-violet-400/60', chip: 'bg-violet-500/15 text-violet-300 border-violet-400/40', dot: '#a78bfa' }, governance: { ring: 'border-fuchsia-400/60', chip: 'bg-fuchsia-500/15 text-fuchsia-300 border-fuchsia-400/40', dot: '#d946ef' }, } const EDGE_COLOR: Record = { generate: '#f59e0b', cdc: '#22d3ee', archive: '#38bdf8', movement: '#a78bfa', mask: '#fb7185', query: '#818cf8', catalog: '#d946ef', } const EDGE_LEGEND: { kind: string; label: string }[] = [ { kind: 'generate', label: 'Generate' }, { kind: 'cdc', label: 'CDC capture' }, { kind: 'archive', label: 'Archive' }, { kind: 'movement', label: 'ETL movement' }, { kind: 'mask', label: 'PII masking' }, { kind: 'query', label: 'Query' }, { kind: 'catalog', label: 'Catalog (OpenMetadata)' }, ] type Anchor = { x: number; y: number; w: number; h: number } function edgePath(a: Anchor, b: Anchor): string { const x1 = a.x const y1 = a.y const x2 = b.x const y2 = b.y const dx = x2 - x1 if (Math.abs(dx) < 8) { const midY = (y1 + y2) / 2 return `M ${x1} ${y1} C ${x1} ${midY}, ${x2} ${midY}, ${x2} ${y2}` } const mx = x1 + dx * 0.5 return `M ${x1} ${y1} C ${mx} ${y1}, ${mx} ${y2}, ${x2} ${y2}` } function fmtDur(s?: number | null): string { if (s == null) return '' if (s < 60) return `${s.toFixed(0)}s` return `${(s / 60).toFixed(1)}m` } /* ── component ──────────────────────────────────────────────────── */ export function DataFlowView() { const [graph, setGraph] = useState(null) const [loading, setLoading] = useState(true) const [refreshing, setRefreshing] = useState(false) const [piiOverlay, setPiiOverlay] = useState(true) const [selected, setSelected] = useState(null) const [triggering, setTriggering] = useState(null) const [etlEnabled, setEtlEnabled] = useState(null) const canvasRef = useRef(null) const nodeRefs = useRef>({}) const [anchors, setAnchors] = useState>({}) const [size, setSize] = useState({ w: 800, h: 460 }) const setNodeRef = useCallback((id: string) => (el: HTMLButtonElement | null) => { nodeRefs.current[id] = el }, []) const load = useCallback(async (refresh = false) => { const g = await fetchDataflow(refresh) if (g) setGraph(g) setLoading(false) }, []) useEffect(() => { load() fetchAgentOpsStatus().then((s) => { const etl = (s as { etl?: { enabled?: boolean } })?.etl if (etl) setEtlEnabled(!!etl.enabled) }) const iv = setInterval(() => load(), 6000) return () => clearInterval(iv) }, [load]) const measure = useCallback(() => { const canvas = canvasRef.current if (!canvas) return const rect = canvas.getBoundingClientRect() if (rect.width < 10 || rect.height < 10) return setSize({ w: rect.width, h: rect.height }) const next: Record = {} for (const [id, el] of Object.entries(nodeRefs.current)) { if (!el) continue const r = el.getBoundingClientRect() next[id] = { x: (r.left + r.right) / 2 - rect.left, y: (r.top + r.bottom) / 2 - rect.top, w: r.width, h: r.height, } } setAnchors(next) }, []) useLayoutEffect(() => { measure() const canvas = canvasRef.current if (!canvas) return const ro = new ResizeObserver(() => measure()) ro.observe(canvas) window.addEventListener('resize', measure) return () => { ro.disconnect() window.removeEventListener('resize', measure) } }, [measure]) useLayoutEffect(() => { measure() }, [measure, graph]) const nodes = graph?.nodes ?? [] const edges = graph?.edges ?? [] const onTrigger = useCallback(async (movementId: string) => { setTriggering(movementId) try { await runDataflowMovement(movementId) setTimeout(() => load(true), 800) } finally { setTimeout(() => setTriggering(null), 1500) } }, [load]) const onToggleEtl = useCallback(async () => { const next = !etlEnabled setEtlEnabled(next) await toggleEtlAgent(next) }, [etlEnabled]) const onRefresh = useCallback(async () => { setRefreshing(true) await load(true) setTimeout(() => setRefreshing(false), 400) }, [load]) const [maskBusy, setMaskBusy] = useState(null) const onToggleMask = useCallback(async (key: string, column: string, masked: boolean) => { setMaskBusy(`${key}.${column}`) try { await setPiiMask(key, column, masked) await load(true) } finally { setMaskBusy(null) } }, [load]) const piiSummary = graph?.pii_summary ?? {} const movementEdges = useMemo( () => edges.filter((e) => e.movement_id), [edges], ) const triggerable = useMemo(() => { const seen = new Set() const out: DataflowEdge[] = [] for (const e of movementEdges) { if (e.movement_id && !seen.has(e.movement_id)) { seen.add(e.movement_id) out.push(e) } } return out }, [movementEdges]) const selNode = nodes.find((n) => n.id === selected) || null return (

Data Flow · live lineage

Generators → sources → CDC/Kafka → lakehouse · click a node for PII detail · trigger movements below

{graph?.cdc?.connected ? 'CDC live' : 'CDC offline'} {graph?.cdc?.window_total ?? 0} chg/15m {(piiSummary.unmasked_columns ?? 0) > 0 ? ( {piiSummary.unmasked_columns} PII unmasked ) : ( {piiSummary.masked_columns ?? 0} PII masked )}
{EDGE_LEGEND.map((l) => ( {l.label} ))}
{/* canvas */}
{loading && (
loading flow…
)} {edges.map((edge, i) => { const a = anchors[edge.from] const b = anchors[edge.to] if (!a || !b) return null const d = edgePath(a, b) const color = EDGE_COLOR[edge.kind] || '#64748b' const running = edge.state === 'running' const active = edge.active || running const dur = 1.6 + (i % 5) * 0.3 return ( {active && ( <> )} ) })} {/* nodes */} {nodes.map((node) => ( setSelected((s) => (s === node.id ? null : node.id))} /> ))} {/* node inspector */} {selNode && (
{selNode.label}

{selNode.sub}

{selNode.metric && (

{selNode.metric}

)} {selNode.url && ( Open in OpenMetadata ↗ )} {selNode.pii?.has_pii ? (
{selNode.pii.all_masked ? : } {selNode.pii.pii_count} PII column{selNode.pii.pii_count === 1 ? '' : 's'} {!selNode.pii.masked_layer && ( )}

Masked columns are hidden from the assistant — it refuses to reveal them.

{selNode.pii.columns.map((c) => { const locked = c.policy_locked const busy = maskBusy === `${selNode.pii!.key}.${c.name}` return ( ) })}
) : (

No PII classified.

)}
)}
{/* movement control strip */}
Data movements
{triggerable.map((e) => { const running = e.state === 'running' const busy = triggering === e.movement_id || running return ( ) })}
) } function NodeCard({ node, piiOverlay, selected, setRef, onClick, }: { node: DataflowNode piiOverlay: boolean selected: boolean setRef: (el: HTMLButtonElement | null) => void onClick: () => void }) { const kind = NODE_KIND[node.kind] || NODE_KIND.source const pii = node.pii const showPii = piiOverlay && pii?.has_pii return ( ) }