From aa9ee66966be238454f553c38cf5ec2724ceb965 Mon Sep 17 00:00:00 2001
From: mo <mo@dell-atc.lan>
Date: Sun, 28 Jun 2026 22:21:24 +0000
Subject: [PATCH] fix: Run control on Data Flow now starts the live data
 generator

Previously Run/Pause/Stop only drove the pulse animation, so on the Data Flow
tab nothing was generated (the streamer is heartbeat-gated and only the Live
dashboard was sending heartbeats). Now:
- New POST /api/federated/live/heartbeat keep-alive endpoint.
- Run seeds an immediate burst into the sources and keeps the generator alive
  (5s heartbeat) while the tab is open; Pause/Stop halts generation.
---
 api/trino_federated.py                      | 13 +++++++
 ui/src/components/features/DataFlowView.tsx | 39 +++++++++++++++++++--
 2 files changed, 49 insertions(+), 3 deletions(-)

diff --git a/api/trino_federated.py b/api/trino_federated.py
index 973375e..d817266 100644
--- a/api/trino_federated.py
+++ b/api/trino_federated.py
@@ -622,6 +622,19 @@ async def toggle_generator(body: dict = Body(default={})):
     return {"ok": True, "enabled": _GEN["enabled"], "interval": _GEN["interval"], "running": _GEN["running"]}
 
 
+@router.post("/live/heartbeat")
+async def live_heartbeat(body: dict = Body(default={})):
+    """Keep-alive for the continuous generator from any watching view (e.g. the
+    Data Flow tab's Run control). active=true → generate; active=false → stop."""
+    active = bool(body.get("active", True))
+    if active:
+        _GEN["enabled"] = True
+        _GEN["last_seen"] = time.time()
+    else:
+        _GEN["last_seen"] = 0.0
+    return {"ok": True, "running": _GEN["running"], "enabled": _GEN["enabled"]}
+
+
 @router.post("/generate")
 async def generate_now(body: dict = Body(default={})):
     """Manual one-shot burst into the source systems (the Data Flow "Generate
diff --git a/ui/src/components/features/DataFlowView.tsx b/ui/src/components/features/DataFlowView.tsx
index dd2153f..420395d 100644
--- a/ui/src/components/features/DataFlowView.tsx
+++ b/ui/src/components/features/DataFlowView.tsx
@@ -206,10 +206,43 @@ export function DataFlowView() {
   }, [genRows, load])
 
   const flowMode = (graph as unknown as { flow?: string })?.flow ?? 'running'
+
+  const heartbeat = useCallback((active: boolean) => {
+    return fetch('/api/federated/live/heartbeat', {
+      method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ active }),
+    }).catch(() => {})
+  }, [])
+
   const onFlow = useCallback(async (action: 'pause' | 'resume' | 'stop') => {
     await setStreamingFlow(action)
-    setTimeout(() => load(true), 300)
-  }, [load])
+    // The master pulse also drives the live data generator: Run starts the
+    // stream into the sources, Pause/Stop halts it.
+    await heartbeat(action === 'resume')
+    if (action === 'resume') {
+      try {
+        const r = await fetch('/api/federated/generate', {
+          method: 'POST', headers: { 'Content-Type': 'application/json' }, body: JSON.stringify({ rows: 800 }),
+        })
+        const d = await r.json()
+        const i = d?.inserted || {}
+        setGenToast(`Generator running — seeded +${(i.orders ?? 0).toLocaleString()} orders, +${(i.telemetry ?? 0).toLocaleString()} telemetry, +${(i.hr_events ?? 0).toLocaleString()} HR, +${(i.supply_events ?? 0).toLocaleString()} supply; now streaming every few seconds via CDC`)
+        setTimeout(() => setGenToast(null), 7000)
+      } catch { /* ignore */ }
+    }
+    setTimeout(() => load(true), 400)
+  }, [load, heartbeat])
+
+  // While the flow is running and this tab is open, keep the generator alive
+  // (it stops a few seconds after you pause/stop or leave the tab).
+  useEffect(() => {
+    if (flowMode !== 'running') {
+      heartbeat(false)
+      return
+    }
+    heartbeat(true)
+    const t = setInterval(() => heartbeat(true), 5000)
+    return () => clearInterval(t)
+  }, [flowMode, heartbeat])
 
   const [maskBusy, setMaskBusy] = useState<string | null>(null)
   const onToggleMask = useCallback(async (key: string, column: string, masked: boolean) => {
@@ -334,7 +367,7 @@ export function DataFlowView() {
             >
               <FileCode2 className="h-3 w-3" /> Scripts
             </button>
-            <div className="inline-flex items-center gap-0.5 rounded border border-border p-0.5" title="Master pulse control">
+            <div className="inline-flex items-center gap-0.5 rounded border border-border p-0.5" title="Master pulse + live generator: Run streams data into the sources, Pause/Stop halts it">
               <button type="button" onClick={() => onFlow('resume')}
                 className={cn('inline-flex items-center gap-1 rounded px-1.5 py-0.5 text-[9px] font-medium transition-colors', flowMode === 'running' ? 'bg-emerald-500/25 text-emerald-200' : 'text-foreground-muted hover:text-foreground')}>
                 <Play className="h-3 w-3" /> Run