From 6f20e24b8b1d2d838d8f4fa9917fed299c6ee2bd Mon Sep 17 00:00:00 2001 From: mo Date: Sat, 27 Jun 2026 02:09:38 +0200 Subject: [PATCH] =?UTF-8?q?feat(ui):=20Data=20Flow=20tab=20=E2=80=94=20liv?= =?UTF-8?q?e=20lineage=20graph=20+=20PII=20overlay=20+=20movement=20trigge?= =?UTF-8?q?rs?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New DataFlowView (topology-style): nodes positioned from /api/dataflow with measured-anchor SVG edges and animated particles (active CDC + running movements). PII overlay shows shield badges + per-column detail inspector. Bottom strip triggers ETL movements and toggles the ETL agent. Wired into SideNav (Data Flow) and App routing; added types + api helpers. --- ui/src/App.tsx | 3 + ui/src/components/features/DataFlowView.tsx | 429 ++++++++++++++++++++ ui/src/components/layout/SideNav.tsx | 3 +- ui/src/lib/api.ts | 34 ++ ui/src/types.ts | 71 ++++ 5 files changed, 539 insertions(+), 1 deletion(-) create mode 100644 ui/src/components/features/DataFlowView.tsx diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 688d885..4bb790a 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -19,6 +19,7 @@ import { StorageView } from './components/features/StorageView' import { HdfsView } from './components/features/HdfsView' import { DataGenView } from './components/features/DataGenView' import { ChangesView } from './components/features/ChangesView' +import { DataFlowView } from './components/features/DataFlowView' import { SearchView } from './components/features/SearchView' import { SshTerminal } from './components/features/SshTerminal' import { TerminalDock } from './components/features/TerminalDock' @@ -131,6 +132,8 @@ export default function App() { cc.setMainView('platform')} /> ) : cc.mainView === 'changes' ? ( + ) : cc.mainView === 'dataflow' ? ( + ) : cc.mainView === 'presentation' ? ( ) : cc.mainView === 'dataquality' ? ( diff --git a/ui/src/components/features/DataFlowView.tsx b/ui/src/components/features/DataFlowView.tsx new file mode 100644 index 0000000..09b44b0 --- /dev/null +++ b/ui/src/components/features/DataFlowView.tsx @@ -0,0 +1,429 @@ +import { useCallback, useEffect, useLayoutEffect, useMemo, useRef, useState } from 'react' +import { GitBranch, Play, RefreshCw, ShieldAlert, ShieldCheck, Loader2, Bot } from 'lucide-react' +import type { DataflowEdge, DataflowGraph, DataflowNode } from '../../types' +import { fetchDataflow, runDataflowMovement, toggleEtlAgent, fetchAgentOpsStatus } 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' }, +} + +const EDGE_COLOR: Record = { + generate: '#f59e0b', + cdc: '#22d3ee', + archive: '#38bdf8', + movement: '#a78bfa', + mask: '#fb7185', + query: '#818cf8', +} + +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' }, +] + +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 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.pii?.has_pii ? ( +
+
+ {selNode.pii.all_masked ? : } + {selNode.pii.pii_count} PII column{selNode.pii.pii_count === 1 ? '' : 's'} +
+
+ {selNode.pii.columns.map((c) => ( +
+ {c.name} + + {c.category} + + {c.masked ? 'masked' : 'raw'} + + +
+ ))} +
+
+ ) : ( +

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 ( + + ) +} diff --git a/ui/src/components/layout/SideNav.tsx b/ui/src/components/layout/SideNav.tsx index fb57301..0dae303 100644 --- a/ui/src/components/layout/SideNav.tsx +++ b/ui/src/components/layout/SideNav.tsx @@ -1,4 +1,4 @@ -import { DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Presentation, Server, TerminalSquare, Cpu, Activity } from 'lucide-react' +import { DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Presentation, Server, TerminalSquare, Cpu, Activity, GitBranch } from 'lucide-react' import type { GpuStatus, WorkloadData } from '../../types' import type { GpuLiveMetrics } from '../../hooks/useLiveMetrics' import { cn } from '../../lib/utils' @@ -26,6 +26,7 @@ const VIEWS: { id: MainView; label: string; icon: typeof LayoutDashboard }[] = [ { id: 'platform', label: 'Data Platform', icon: LayoutDashboard }, { id: 'datagen', label: 'Data Generation', icon: Cpu }, { id: 'changes', label: 'Live Changes', icon: Activity }, + { id: 'dataflow', label: 'Data Flow', icon: GitBranch }, { id: 'presentation', label: 'Presentation', icon: Presentation }, { id: 'dataquality', label: 'Data Quality', icon: DatabaseZap }, { id: 'knowledge', label: 'Knowledge Chat', icon: MessageSquare }, diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index 61b7945..8a66385 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -4,8 +4,11 @@ import type { Approval, CdcChange, CdcStats, + DataflowGraph, FeedEntry, GpuStatus, + Movement, + PiiDataset, StatusData, TerminalLine, WorkloadData, @@ -129,6 +132,37 @@ export function runAgentOpOnce(source?: string, op?: string) { }) } +export async function fetchDataflow(refresh = false): Promise { + return fetchJson(`/api/dataflow${refresh ? '?refresh=true' : ''}`, 25000) +} + +export function runDataflowMovement(movementId: string, conf?: Record) { + return fetch(`/api/dataflow/${movementId}/run`, { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(conf ? { conf } : {}), + }) +} + +export async function fetchMovements(): Promise { + const j = await fetchJson<{ movements?: Movement[] }>('/api/movements', 8000) + return j?.movements || [] +} + +export async function fetchPii(refresh = false): Promise<{ datasets: PiiDataset[]; summary: Record } | null> { + return fetchJson<{ datasets: PiiDataset[]; summary: Record }>( + `/api/pii${refresh ? '?refresh=true' : ''}`, 15000, + ) +} + +export function toggleEtlAgent(enabled?: boolean) { + return fetch('/api/agent-ops/etl/toggle', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(enabled === undefined ? {} : { enabled }), + }) +} + export async function decideApproval(id: string, approved: boolean, decidedBy: string, note: string) { return fetch(`/api/approvals/${id}/decide`, { method: 'POST', diff --git a/ui/src/types.ts b/ui/src/types.ts index 9481b1b..83c08c7 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -70,6 +70,77 @@ export type CdcStats = { consumed: number } +export type PiiColumn = { name: string; category: string; masked: boolean } + +export type DataflowNode = { + id: string + label: string + sub: string + kind: string + x: number + y: number + level: string + metric: string | null + pii?: { + has_pii: boolean + pii_count: number + all_masked: boolean + masked_layer: boolean + categories: string[] + columns: PiiColumn[] + } +} + +export type DataflowEdge = { + from: string + to: string + kind: string + movement_id?: string + state?: string + last_rows?: number | null + last_duration_s?: number | null + active?: boolean +} + +export type DataflowGraph = { + ok: boolean + nodes: DataflowNode[] + edges: DataflowEdge[] + pii_summary: { datasets?: number; pii_columns?: number; masked_columns?: number; unmasked_columns?: number } + cdc: { connected?: boolean; consumed?: number; window_total?: number } + ts: number +} + +export type Movement = { + id: string + label: string + kind: string + dag_id: string + agent: string + from: string + to: string + last_run?: { + state?: string + rows?: number | null + duration_s?: number | null + ended_at?: string + started_at?: string + } | null +} + +export type PiiDataset = { + key: string + node_id: string + label: string + table: string + exists: boolean + masked_layer: boolean + pii_columns: PiiColumn[] + pii_count: number + has_pii: boolean + all_masked: boolean +} + export type DomainStatus = { level: 'ok' | 'warn' | 'down' | 'unknown' label: string