From 921342442f101c57a4b5b13b42d8fb38d0ba0d59 Mon Sep 17 00:00:00 2001 From: mo Date: Sat, 27 Jun 2026 01:44:26 +0200 Subject: [PATCH] =?UTF-8?q?feat(ui):=20Live=20Changes=20(CDC)=20tab=20?= =?UTF-8?q?=E2=80=94=20real-time=20Debezium=20stream=20with=20filters,=20v?= =?UTF-8?q?olume=20bars=20and=20before/after=20diff?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- ui/src/App.tsx | 3 + ui/src/components/features/ChangesView.tsx | 227 +++++++++++++++++++++ ui/src/components/layout/SideNav.tsx | 5 +- ui/src/hooks/useCommandCenter.ts | 6 +- ui/src/lib/api.ts | 37 ++++ ui/src/types.ts | 24 +++ 6 files changed, 299 insertions(+), 3 deletions(-) create mode 100644 ui/src/components/features/ChangesView.tsx diff --git a/ui/src/App.tsx b/ui/src/App.tsx index 17050c7..688d885 100644 --- a/ui/src/App.tsx +++ b/ui/src/App.tsx @@ -18,6 +18,7 @@ import { KnowledgeChatView } from './components/features/KnowledgeChatView' 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 { SearchView } from './components/features/SearchView' import { SshTerminal } from './components/features/SshTerminal' import { TerminalDock } from './components/features/TerminalDock' @@ -128,6 +129,8 @@ export default function App() { /> ) : cc.mainView === 'datagen' ? ( cc.setMainView('platform')} /> + ) : cc.mainView === 'changes' ? ( + ) : cc.mainView === 'presentation' ? ( ) : cc.mainView === 'dataquality' ? ( diff --git a/ui/src/components/features/ChangesView.tsx b/ui/src/components/features/ChangesView.tsx new file mode 100644 index 0000000..5af3563 --- /dev/null +++ b/ui/src/components/features/ChangesView.tsx @@ -0,0 +1,227 @@ +import { useCallback, useEffect, useMemo, useState } from 'react' +import { Activity, Radio, RefreshCw } from 'lucide-react' +import { fetchChanges, fetchChangeStats } from '../../lib/api' +import type { CdcChange, CdcStats } from '../../types' +import { cn } from '../../lib/utils' + +const OP_STYLE: Record = { + insert: { label: 'INSERT', cls: 'bg-emerald-500/15 text-emerald-300 border-emerald-500/30' }, + update: { label: 'UPDATE', cls: 'bg-amber-500/15 text-amber-300 border-amber-500/30' }, + delete: { label: 'DELETE', cls: 'bg-rose-500/15 text-rose-300 border-rose-500/30' }, + snapshot: { label: 'SNAPSHOT', cls: 'bg-sky-500/15 text-sky-300 border-sky-500/30' }, +} + +const SOURCE_COLOR: Record = { + postgres: '#38bdf8', + mysql: '#f59e0b', + mongodb: '#34d399', + cassandra: '#a78bfa', + neo4j: '#f472b6', +} + +const SOURCES = ['all', 'postgres', 'mysql', 'mongodb', 'cassandra', 'neo4j'] +const OPS = ['all', 'insert', 'update', 'delete'] + +function opOf(o: string) { + return OP_STYLE[o] || { label: o.toUpperCase(), cls: 'bg-slate-500/15 text-slate-300 border-slate-500/30' } +} + +function timeAgo(ts: string) { + const d = Date.now() - new Date(ts).getTime() + if (d < 1000) return 'now' + if (d < 60000) return `${Math.floor(d / 1000)}s ago` + if (d < 3600000) return `${Math.floor(d / 60000)}m ago` + return `${Math.floor(d / 3600000)}h ago` +} + +function Diff({ change }: { change: CdcChange }) { + const keys = useMemo(() => { + const set = new Set() + for (const k of Object.keys(change.before || {})) set.add(k) + for (const k of Object.keys(change.after || {})) set.add(k) + return Array.from(set).filter((k) => k !== 'payload').slice(0, 24) + }, [change]) + const fmt = (v: unknown) => { + if (v === null || v === undefined) return '∅' + if (typeof v === 'object') return JSON.stringify(v).slice(0, 60) + return String(v).slice(0, 60) + } + return ( +
+ + + + + + + + + + {keys.map((k) => { + const b = (change.before || {})[k] + const a = (change.after || {})[k] + const changed = JSON.stringify(b) !== JSON.stringify(a) + return ( + + + + + + ) + })} + +
columnbeforeafter
{k}{fmt(b)}{fmt(a)}
+
+ ) +} + +export function ChangesView({ liveChanges }: { liveChanges: CdcChange[] }) { + const [seed, setSeed] = useState([]) + const [stats, setStats] = useState(null) + const [source, setSource] = useState('all') + const [op, setOp] = useState('all') + const [expanded, setExpanded] = useState(null) + const [connected, setConnected] = useState(false) + + const load = useCallback(async () => { + const [c, s] = await Promise.all([fetchChanges({ limit: 150 }), fetchChangeStats(15)]) + setSeed(c.changes) + setConnected(c.connected) + if (s) setStats(s) + }, []) + + useEffect(() => { + load() + const iv = setInterval(() => fetchChangeStats(15).then((s) => s && setStats(s)), 5000) + return () => clearInterval(iv) + }, [load]) + + // Merge live (WS) with seeded backlog, dedupe by id, newest first. + const merged = useMemo(() => { + const byId = new Map() + for (const c of liveChanges) byId.set(c.id, c) + for (const c of seed) if (!byId.has(c.id)) byId.set(c.id, c) + return Array.from(byId.values()).sort((x, y) => (y.ts > x.ts ? 1 : -1)) + }, [liveChanges, seed]) + + const filtered = useMemo( + () => merged.filter((c) => (source === 'all' || c.source === source) && (op === 'all' || c.op === op)).slice(0, 200), + [merged, source, op], + ) + + const maxBucket = Math.max(1, ...(stats?.buckets || []).map((b) => b.n)) + + return ( +
+ {/* Header */} +
+
+

+ Live Changes · CDC Stream +

+

+ Real-time Debezium change data capture from all source databases via Kafka +

+
+
+ + {connected ? 'STREAMING' : 'OFFLINE'} + + +
+
+ + {/* Stat cards */} +
+
+
Changes / 15 min
+
{stats?.total ?? 0}
+
+
+
Total consumed
+
{stats?.consumed ?? 0}
+
+
+
By operation
+
+ {Object.entries(stats?.by_op || {}).map(([k, v]) => ( + {opOf(k).label} {v} + ))} + {!Object.keys(stats?.by_op || {}).length && —} +
+
+
+
By source
+
+ {Object.entries(stats?.by_source || {}).map(([k, v]) => ( + + {k} {v} + + ))} + {!Object.keys(stats?.by_source || {}).length && —} +
+
+
+ + {/* Volume sparkbars */} +
+
Change volume per minute (last 15m)
+
+ {(stats?.buckets || []).map((b) => ( +
+
+
+ ))} + {!(stats?.buckets || []).length &&
No changes in the window yet…
} +
+
+ + {/* Filters */} +
+ Source + {SOURCES.map((s) => ( + + ))} + Op + {OPS.map((o) => ( + + ))} +
+ + {/* Live list */} +
+ {filtered.length === 0 && ( +
+ Waiting for changes… trigger data generation or agent DML to see live CDC events. +
+ )} + {filtered.map((c) => ( +
+ + {expanded === c.id &&
} +
+ ))} +
+
+ ) +} diff --git a/ui/src/components/layout/SideNav.tsx b/ui/src/components/layout/SideNav.tsx index 559c5de..fb57301 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 } from 'lucide-react' +import { DatabaseZap, HardDrive, Search, LayoutDashboard, MessageSquare, Presentation, Server, TerminalSquare, Cpu, Activity } from 'lucide-react' import type { GpuStatus, WorkloadData } from '../../types' import type { GpuLiveMetrics } from '../../hooks/useLiveMetrics' import { cn } from '../../lib/utils' @@ -6,7 +6,7 @@ import { viewTabActive, viewTabIdle } from '../../lib/tabActive' import { GpuMatrixPanel } from '../features/GpuMatrixPanel' import { LabHealthPanel } from '../features/LabHealthPanel' -type MainView = 'platform' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'datagen' | 'approvals' +type MainView = 'platform' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'datagen' | 'approvals' | 'changes' | 'dataflow' type Props = { workload: WorkloadData | null @@ -25,6 +25,7 @@ type Props = { 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: 'presentation', label: 'Presentation', icon: Presentation }, { id: 'dataquality', label: 'Data Quality', icon: DatabaseZap }, { id: 'knowledge', label: 'Knowledge Chat', icon: MessageSquare }, diff --git a/ui/src/hooks/useCommandCenter.ts b/ui/src/hooks/useCommandCenter.ts index e25f163..ba3e15c 100644 --- a/ui/src/hooks/useCommandCenter.ts +++ b/ui/src/hooks/useCommandCenter.ts @@ -19,6 +19,7 @@ import type { Agent, AgentAnim, Approval, + CdcChange, ChatMessage, FeedEntry, GpuStatus, @@ -68,7 +69,8 @@ export function useCommandCenter() { const [selectedNode, setSelectedNode] = useState(null) const [nodeDetail, setNodeDetail] = useState(null) const [nodeBusy, setNodeBusy] = useState(false) - const [mainView, setMainView] = useState<'platform' | 'approvals' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'datagen'>('platform') + const [mainView, setMainView] = useState<'platform' | 'approvals' | 'presentation' | 'dataquality' | 'knowledge' | 'storage' | 'hdfs' | 'search' | 'datagen' | 'changes' | 'dataflow'>('platform') + const [changes, setChanges] = useState([]) const [genPulse, setGenPulse] = useState(false) const genPulseTimer = useRef | null>(null) const pulseFlow = useCallback(() => { @@ -135,6 +137,7 @@ export function useCommandCenter() { if (msg.type === 'terminal') appendTerminal(msg.line) if (msg.type === 'terminal_history' && msg.terminals) setTerminals(msg.terminals) if (msg.type === 'feed') setFeed((prev) => [msg.entry, ...prev].slice(0, 100)) + if (msg.type === 'cdc_change' && msg.entry) setChanges((prev) => [msg.entry, ...prev].slice(0, 400)) if (msg.type === 'agent_dispatch') { setSelectedAgentId(msg.agent_id) setAnims((p) => ({ ...p, [msg.agent_id]: { agentId: msg.agent_id, state: 'walk', zone: msg.zone } })) @@ -343,6 +346,7 @@ export function useCommandCenter() { nodeBusy, mainView, setMainView, + changes, genPulse, pulseFlow, approvalHighlight, diff --git a/ui/src/lib/api.ts b/ui/src/lib/api.ts index 8263793..61b7945 100644 --- a/ui/src/lib/api.ts +++ b/ui/src/lib/api.ts @@ -2,6 +2,8 @@ import type { PresentationData, Agent, Approval, + CdcChange, + CdcStats, FeedEntry, GpuStatus, StatusData, @@ -92,6 +94,41 @@ export async function fetchPresentation(): Promise { return fetchJson('/api/presentation', 60000) } +export async function fetchChanges(opts: { source?: string; op?: string; limit?: number } = {}) { + const p = new URLSearchParams() + if (opts.source) p.set('source', opts.source) + if (opts.op) p.set('op', opts.op) + p.set('limit', String(opts.limit ?? 150)) + const j = await fetchJson<{ changes?: CdcChange[]; connected?: boolean; consumed?: number }>( + `/api/changes?${p.toString()}`, 8000, + ) + return { changes: j?.changes || [], connected: !!j?.connected, consumed: j?.consumed || 0 } +} + +export async function fetchChangeStats(minutes = 15): Promise { + return fetchJson(`/api/changes/stats?minutes=${minutes}`, 8000) +} + +export async function fetchAgentOpsStatus() { + return fetchJson>('/api/agent-ops/status', 8000) +} + +export function toggleAgentOps(enabled?: boolean) { + return fetch('/api/agent-ops/toggle', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(enabled === undefined ? {} : { enabled }), + }) +} + +export function runAgentOpOnce(source?: string, op?: string) { + return fetch('/api/agent-ops/run-once', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ source, op }), + }) +} + 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 391d209..9481b1b 100644 --- a/ui/src/types.ts +++ b/ui/src/types.ts @@ -46,6 +46,30 @@ export type FeedEntry = { level: string } +export type CdcChange = { + id: string + ts: string + source: string + table: string + topic: string + op: string + summary: string + before: Record | null + after: Record | null +} + +export type CdcStats = { + ok: boolean + window_minutes: number + total: number + by_source: Record + by_op: Record + by_table: Record + buckets: { t: string; n: number }[] + connected: boolean + consumed: number +} + export type DomainStatus = { level: 'ok' | 'warn' | 'down' | 'unknown' label: string