Merge pull request #612 from tapdata/TAP-12260-data-trace-dev

fix(TAP-12260): [T3 to dev]The traceability interface returns data, but the fron…

Author: feynmx

packages/api/src/core/data-trace-stream-parser.ts

@@ -0,0 +1,51 @@
+export interface TraceStreamProcessor {
+  (text: string): void
+  flush: () => void
+}
+
+export function createTraceStreamProcessor(
+  onTraceValue: (nodeId: string, data: Record<string, any>) => void,
+): TraceStreamProcessor {
+  let lastIndex = 0
+  let pendingLine = ''
+
+  function parseLine(line: string) {
+    const trimmed = line.trim()
+    if (!trimmed) return
+
+    const parsed = JSON.parse(trimmed)
+    if (parsed.nodeId !== undefined && parsed.type === 'TRACE_VALUE') {
+      onTraceValue(parsed.nodeId, parsed)
+    }
+  }
+
+  const processChunk = ((text: string) => {
+    const newText = text.slice(lastIndex)
+    lastIndex = text.length
+
+    const lines = (pendingLine + newText).split('\n')
+    pendingLine = lines.pop() ?? ''
+
+    for (const line of lines) {
+      try {
+        parseLine(line)
+      } catch {
+        // Ignore malformed non-trace lines from the stream.
+      }
+    }
+  }) as TraceStreamProcessor
+
+  processChunk.flush = () => {
+    if (!pendingLine.trim()) return
+
+    try {
+      parseLine(pendingLine)
+    } catch {
+      // Ignore an incomplete final line.
+    } finally {
+      pendingLine = ''
+    }
+  }
+
+  return processChunk
+}

packages/api/src/core/data-trace.ts

@@ -1,4 +1,5 @@
 import { requestClient } from '../request'
+import { createTraceStreamProcessor } from './data-trace-stream-parser'
 
 const BASE_URL = '/api/lineage/wide-table'
 
@@ -98,27 +99,9 @@ export function getTraceData(
   callbacks: TraceStreamCallbacks,
 ): AbortController {
   const controller = new AbortController()
-  let lastIndex = 0
-
-  function processChunk(text: string) {
-    const newText = text.slice(lastIndex)
-    lastIndex = text.length
-
-    const lines = newText.split('\n')
-    for (const line of lines) {
-      const trimmed = line.trim()
-      if (!trimmed) continue
-      try {
-        const parsed = JSON.parse(trimmed)
-        if (parsed.nodeId !== undefined && parsed.type === 'TRACE_VALUE') {
-          console.log('parsed', parsed)
-          callbacks.onNodeData?.(parsed.nodeId, parsed)
-        }
-      } catch {
-        // ignore incomplete line (will be completed in next chunk)
-      }
-    }
-  }
+  const processChunk = createTraceStreamProcessor((nodeId, parsed) => {
+    callbacks.onNodeData?.(nodeId, parsed)
+  })
 
   requestClient
     .post(`${BASE_URL}/trace/stream`, data, {
@@ -135,6 +118,7 @@ export function getTraceData(
       },
     })
     .then(() => {
+      processChunk.flush()
       callbacks.onDone?.()
     })
     .catch((error: any) => {