fix: stabilize datasource progress and websocket flow

2026-04-01 12:04:28 +08:00
parent 126d4dadb7
commit b8f70f8b71
11 changed files with 408 additions and 134 deletions
--- a/frontend/src/pages/BGP/BGP.tsx
+++ b/frontend/src/pages/BGP/BGP.tsx
@@ -110,7 +110,6 @@ function BGP() {
  const [events, setEvents] = useState<BGPEvent[]>([])
  const [collectors, setCollectors] = useState<BGPCollectorCoverage[]>([])
  const [incidentSummary, setIncidentSummary] = useState<Summary | null>(null)
-  const [anomalySummary, setAnomalySummary] = useState<Summary | null>(null)
  const [eventSummary, setEventSummary] = useState<EventSummary | null>(null)
  const [collectorSummary, setCollectorSummary] = useState<CollectorSummary | null>(null)

@@ -118,11 +117,10 @@ function BGP() {
    const load = async () => {
      setLoading(true)
      try {
-        const [incidentsRes, incidentSummaryRes, anomaliesRes, anomalySummaryRes, eventsRes, eventSummaryRes, collectorsRes, collectorSummaryRes] = await Promise.all([
+        const [incidentsRes, incidentSummaryRes, anomaliesRes, eventsRes, eventSummaryRes, collectorsRes, collectorSummaryRes] = await Promise.all([
          axios.get('/api/v1/bgp/incidents', { params: { page_size: 50 } }),
          axios.get('/api/v1/bgp/incidents/summary'),
          axios.get('/api/v1/bgp/anomalies', { params: { page_size: 100 } }),
-          axios.get('/api/v1/bgp/anomalies/summary'),
          axios.get('/api/v1/bgp/events', { params: { page_size: 20 } }),
          axios.get('/api/v1/bgp/events/summary'),
          axios.get('/api/v1/bgp/collectors'),
@@ -131,7 +129,6 @@ function BGP() {
        setIncidents(incidentsRes.data.data || [])
        setIncidentSummary(incidentSummaryRes.data)
        setAnomalies(anomaliesRes.data.data || [])
-        setAnomalySummary(anomalySummaryRes.data)
        setEvents(eventsRes.data.data || [])
        setEventSummary(eventSummaryRes.data)
        setCollectors(collectorsRes.data.data || [])
--- a/frontend/src/pages/Dashboard/Dashboard.tsx
+++ b/frontend/src/pages/Dashboard/Dashboard.tsx
@@ -14,6 +14,7 @@ import { Link } from 'react-router-dom'
 import axios from 'axios'
 import { useAuthStore } from '../../stores/auth'
 import AppLayout from '../../components/AppLayout/AppLayout'
+import { useWebSocket } from '../../hooks/useWebSocket'
 import { formatDateTimeZhCN } from '../../utils/datetime'

 const { Title, Text } = Typography
@@ -150,54 +151,21 @@ function Dashboard() {
    fetchStats()
  }, [token, clearAuth])

-  useEffect(() => {
-    if (!token) return
-
-    let ws: WebSocket | null = null
-    let reconnectTimer: ReturnType<typeof setTimeout> | null = null
-
-    const connectWs = () => {
-      try {
-        ws = new WebSocket(`ws://localhost:8000/ws?token=${token}`)
-
-        ws.onopen = () => {
-          setWsConnected(true)
-          ws?.send(JSON.stringify({ type: 'subscribe', data: { channels: ['dashboard'] } }))
-        }
-
-        ws.onmessage = (event) => {
-          try {
-            const msg = JSON.parse(event.data)
-            if (msg.type === 'data_frame' && msg.channel === 'dashboard') {
-              const nextStats = msg.payload?.stats as Stats
-              cachedDashboardStats = nextStats
-              setStats(nextStats)
-            }
-          } catch (e) {
-            console.error('Parse WS message error:', e)
-          }
-        }
-
-        ws.onclose = () => {
-          setWsConnected(false)
-          reconnectTimer = setTimeout(connectWs, 3000)
-        }
-
-        ws.onerror = () => {
-          setWsConnected(false)
-        }
-      } catch (e) {
-        console.error('WS connect error:', e)
+  const { connected: dashboardSocketConnected } = useWebSocket({
+    autoConnect: true,
+    autoSubscribe: ['dashboard'],
+    onMessage: (message) => {
+      if (message.type === 'data_frame' && message.channel === 'dashboard' && message.payload?.stats) {
+        const nextStats = message.payload.stats as Stats
+        cachedDashboardStats = nextStats
+        setStats(nextStats)
      }
-    }
+    },
+  })

-    connectWs()
-
-    return () => {
-      ws?.close()
-      if (reconnectTimer) clearTimeout(reconnectTimer)
-    }
-  }, [token])
+  useEffect(() => {
+    setWsConnected(dashboardSocketConnected)
+  }, [dashboardSocketConnected])

  const handleRetry = () => {
    window.location.reload()
@@ -376,8 +344,8 @@ function Dashboard() {

  if (loading && !stats) {
    return (
-      <div style={{ display: 'flex', justifyContent: 'center', alignItems: 'center', height: '100vh' }}>
-        <Spin size="large" tip="加载中..." />
+      <div style={{ height: '100vh' }}>
+        <Spin size="large" tip="加载中..." fullscreen />
      </div>
    )
  }
--- a/frontend/src/pages/DataSources/DataSources.tsx
+++ b/frontend/src/pages/DataSources/DataSources.tsx
@@ -1,6 +1,6 @@
 import { useCallback, useEffect, useRef, useState } from 'react'
 import {
-  Table, Tag, Space, message, Button, Form, Input, Select, Progress, Checkbox,
+  Table, Tag, Space, Button, Form, Input, Select, Progress, Checkbox, message,
  Drawer, Tabs, Empty, Tooltip, Popconfirm, Collapse, InputNumber, Row, Col, Card
 } from 'antd'
 import {
@@ -47,6 +47,37 @@ interface TaskTrackerState {
  error_message?: string | null
 }

+interface BulkProgressItem {
+  task_id: number | null
+  progress: number
+  status: string | null
+  phase: string | null
+  is_running: boolean
+}
+
+interface BulkProgressBatch {
+  sourceIds: number[]
+  items: Record<number, BulkProgressItem>
+}
+
+function finalizeBulkProgressBatch(batch: BulkProgressBatch | null): BulkProgressBatch | null {
+  if (!batch || batch.sourceIds.length === 0) {
+    return null
+  }
+
+  const hasRunningItem = batch.sourceIds.some((sourceId) => batch.items[sourceId]?.is_running)
+  if (hasRunningItem) {
+    return batch
+  }
+
+  const allFinished = batch.sourceIds.every((sourceId) => {
+    const status = batch.items[sourceId]?.status
+    return Boolean(status && status !== 'running')
+  })
+
+  return allFinished ? null : batch
+}
+
 interface WebSocketTaskMessage {
  type: string
  channel?: string
@@ -90,6 +121,7 @@ interface ViewDataSource {
 }

 function DataSources() {
+  const [messageApi, contextHolder] = message.useMessage()
  const [activeTab, setActiveTab] = useState('builtin')
  const [builtInSources, setBuiltInSources] = useState<BuiltInDataSource[]>([])
  const [customSources, setCustomSources] = useState<CustomDataSource[]>([])
@@ -126,6 +158,7 @@ function DataSources() {
  }, [])

  const [taskProgress, setTaskProgress] = useState<Record<number, TaskTrackerState>>({})
+  const [bulkProgressBatch, setBulkProgressBatch] = useState<BulkProgressBatch | null>(null)
  const activeBuiltInCount = builtInSources.filter((source) => source.is_active).length
  const runningBuiltInCount = builtInSources.filter((source) => {
    const trackedTask = taskProgress[source.id]
@@ -135,13 +168,42 @@ function DataSources() {
    const trackedTask = taskProgress[source.id]
    return trackedTask?.is_running || source.is_running
  })
-  const aggregateProgress = runningBuiltInSources.length > 0
+  const aggregateProgress = bulkProgressBatch && bulkProgressBatch.sourceIds.length > 0
    ? Math.round(
-        runningBuiltInSources.reduce((sum, source) => {
-          const trackedTask = taskProgress[source.id]
-          return sum + (trackedTask?.progress ?? source.progress ?? 0)
-        }, 0) / runningBuiltInSources.length
+        bulkProgressBatch.sourceIds.reduce((sum, sourceId) => {
+          const item = bulkProgressBatch.items[sourceId]
+          if (!item) return sum
+          if (item.status && item.status !== 'running') {
+            return sum + 100
+          }
+          return sum + (item.progress || 0)
+        }, 0) / bulkProgressBatch.sourceIds.length
      )
+    : runningBuiltInSources.length > 0
+      ? Math.round(
+          runningBuiltInSources.reduce((sum, source) => {
+            const trackedTask = taskProgress[source.id]
+            return sum + (trackedTask?.progress ?? source.progress ?? 0)
+          }, 0) / runningBuiltInSources.length
+        )
+      : 0
+
+  const bulkBatchRunningCount = bulkProgressBatch
+    ? bulkProgressBatch.sourceIds.filter((sourceId) => bulkProgressBatch.items[sourceId]?.is_running).length
+    : 0
+
+  const bulkBatchSuccessCount = bulkProgressBatch
+    ? bulkProgressBatch.sourceIds.filter((sourceId) => {
+        const status = bulkProgressBatch.items[sourceId]?.status
+        return status === 'success'
+      }).length
+    : 0
+
+  const bulkBatchFailedCount = bulkProgressBatch
+    ? bulkProgressBatch.sourceIds.filter((sourceId) => {
+        const status = bulkProgressBatch.items[sourceId]?.status
+        return Boolean(status && status !== 'running' && status !== 'success')
+      }).length
    : 0

  const handleTaskSocketMessage = useCallback((message: WebSocketTaskMessage) => {
@@ -151,6 +213,9 @@ function DataSources() {

    const payload = message.payload
    const sourceId = payload.datasource_id
+    if (typeof sourceId !== 'number') {
+      return
+    }
    const nextState: TaskTrackerState = {
      task_id: payload.task_id ?? null,
      progress: payload.progress ?? 0,
@@ -175,6 +240,28 @@ function DataSources() {
      return next
    })

+    setBulkProgressBatch((prev) => {
+      if (!prev || !prev.sourceIds.includes(sourceId)) {
+        return prev
+      }
+
+      const nextItems = {
+        ...prev.items,
+        [sourceId]: {
+          task_id: payload.task_id ?? prev.items[sourceId]?.task_id ?? null,
+          progress: payload.status && payload.status !== 'running' ? 100 : (payload.progress ?? prev.items[sourceId]?.progress ?? 0),
+          is_running: payload.status === 'running',
+          phase: payload.phase ?? prev.items[sourceId]?.phase ?? null,
+          status: payload.status ?? prev.items[sourceId]?.status ?? null,
+        },
+      }
+
+      return finalizeBulkProgressBatch({
+        ...prev,
+        items: nextItems,
+      })
+    })
+
    if (payload.status && payload.status !== 'running') {
      void fetchData()
    }
@@ -265,6 +352,28 @@ function DataSources() {
        return next
      })

+      setBulkProgressBatch((prev) => {
+        if (!prev) return prev
+
+        const nextItems = { ...prev.items }
+        for (const [sourceId, state] of Object.entries(updates)) {
+          const numericSourceId = Number(sourceId)
+          if (!prev.sourceIds.includes(numericSourceId)) continue
+          nextItems[numericSourceId] = {
+            task_id: state.task_id,
+            progress: state.status && state.status !== 'running' ? 100 : state.progress,
+            is_running: state.is_running,
+            phase: state.phase ?? null,
+            status: state.status ?? null,
+          }
+        }
+
+        return finalizeBulkProgressBatch({
+          ...prev,
+          items: nextItems,
+        })
+      })
+
      if (Object.values(updates).some((state) => !state.is_running)) {
        fetchData()
      }
@@ -276,7 +385,7 @@ function DataSources() {
  const handleTrigger = async (id: number) => {
    try {
      const res = await axios.post(`/api/v1/datasources/${id}/trigger`)
-      message.success('任务已触发')
+      messageApi.success('任务已触发')
      if (res.data.task_id) {
        setTaskProgress(prev => ({
          ...prev,
@@ -296,7 +405,7 @@ function DataSources() {
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '触发失败')
+      messageApi.error(err.response?.data?.detail || '触发失败')
    }
  }

@@ -313,6 +422,22 @@ function DataSources() {
      const skippedOther = skipped.filter((item: { reason?: string }) => item.reason !== 'within_frequency_window')

      if (triggered.length > 0) {
+        setBulkProgressBatch({
+          sourceIds: triggered.map((item: { id: number }) => item.id),
+          items: Object.fromEntries(
+            triggered.map((item: { id: number; task_id?: number | null }) => [
+              item.id,
+              {
+                task_id: item.task_id ?? null,
+                progress: 0,
+                is_running: true,
+                phase: 'queued',
+                status: 'running',
+              } satisfies BulkProgressItem,
+            ])
+          ),
+        })
+
        setTaskProgress((prev) => {
          const next = { ...prev }
          for (const item of triggered) {
@@ -336,11 +461,11 @@ function DataSources() {
        failed.length > 0 ? `失败 ${failed.length} 个` : null,
      ].filter(Boolean)

-      message.success(summaryParts.join('，'))
+      messageApi.success(summaryParts.join('，'))
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '全触发失败')
+      messageApi.error(err.response?.data?.detail || '全触发失败')
    } finally {
      setTriggerAllLoading(false)
    }
@@ -350,11 +475,11 @@ function DataSources() {
    const endpoint = current ? 'disable' : 'enable'
    try {
      await axios.post(`/api/v1/datasources/${id}/${endpoint}`)
-      message.success(`${current ? '已禁用' : '已启用'}`)
+      messageApi.success(`${current ? '已禁用' : '已启用'}`)
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '操作失败')
+      messageApi.error(err.response?.data?.detail || '操作失败')
    }
  }

@@ -362,12 +487,12 @@ function DataSources() {
    if (!viewingSource) return
    try {
      const res = await axios.delete(`/api/v1/datasources/${viewingSource.id}/data`)
-      message.success(res.data.message || '数据已删除')
+      messageApi.success(res.data.message || '数据已删除')
      setViewDrawerVisible(false)
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '删除数据失败')
+      messageApi.error(err.response?.data?.detail || '删除数据失败')
    }
  }

@@ -395,7 +520,7 @@ function DataSources() {
      setRecordCount(statsRes.data.total_records || 0)
      setViewDrawerVisible(true)
    } catch (error) {
-      message.error('获取数据源信息失败')
+      messageApi.error('获取数据源信息失败')
    }
  }

@@ -403,11 +528,11 @@ function DataSources() {
    if (!viewingSource) return
    try {
      await axios.post(`/api/v1/datasources/${viewingSource.id}/trigger`)
-      message.success('已触发更新')
+      messageApi.success('已触发更新')
      setViewDrawerVisible(false)
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '更新失败')
+      messageApi.error(err.response?.data?.detail || '更新失败')
    }
  }

@@ -419,13 +544,13 @@ function DataSources() {
      const res = await axios.post('/api/v1/datasources/configs/test', values)
      setTestResult(res.data)
      if (res.data.success) {
-        message.success('连接测试成功')
+        messageApi.success('连接测试成功')
      } else {
-        message.error('连接测试失败')
+        messageApi.error('连接测试失败')
      }
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string; message?: string } } }
-      message.error(err.response?.data?.message || err.response?.data?.detail || '测试失败')
+      messageApi.error(err.response?.data?.message || err.response?.data?.detail || '测试失败')
    } finally {
      setTesting(false)
    }
@@ -436,10 +561,10 @@ function DataSources() {
      const values = await form.validateFields()
      if (editingConfig) {
        await axios.put(`/api/v1/datasources/configs/${editingConfig.id}`, values)
-        message.success('配置已更新')
+        messageApi.success('配置已更新')
      } else {
        await axios.post('/api/v1/datasources/configs', values)
-        message.success('配置已创建')
+        messageApi.success('配置已创建')
      }
      setDrawerVisible(false)
      form.resetFields()
@@ -448,29 +573,29 @@ function DataSources() {
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string; message?: string } } }
-      message.error(err.response?.data?.message || err.response?.data?.detail || '保存失败')
+      messageApi.error(err.response?.data?.message || err.response?.data?.detail || '保存失败')
    }
  }

  const handleDelete = async (id: number) => {
    try {
      await axios.delete(`/api/v1/datasources/configs/${id}`)
-      message.success('配置已删除')
+      messageApi.success('配置已删除')
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '删除失败')
+      messageApi.error(err.response?.data?.detail || '删除失败')
    }
  }

  const handleToggleCustom = async (id: number, current: boolean) => {
    try {
      await axios.put(`/api/v1/datasources/configs/${id}`, { is_active: !current })
-      message.success(`${current ? '已禁用' : '已启用'}`)
+      messageApi.success(`${current ? '已禁用' : '已启用'}`)
      fetchData()
    } catch (error: unknown) {
      const err = error as { response?: { data?: { detail?: string } } }
-      message.error(err.response?.data?.detail || '操作失败')
+      messageApi.error(err.response?.data?.detail || '操作失败')
    }
  }

@@ -509,9 +634,9 @@ function DataSources() {
        document.execCommand('copy')
        document.body.removeChild(textArea)
      }
-      message.success(successText)
+      messageApi.success(successText)
    } catch {
-      message.error('复制失败，请手动复制')
+      messageApi.error('复制失败，请手动复制')
    }
  }

@@ -719,9 +844,30 @@ function DataSources() {
                />
              </div>
              <div className="data-source-bulk-toolbar__stats">
-                <Tag color="blue">内置 {builtInSources.length}</Tag>
-                <Tag color="green">已启用 {activeBuiltInCount}</Tag>
-                <Tag color="processing">执行中 {runningBuiltInCount}</Tag>
+                <div className="data-source-bulk-toolbar__stat-pill">
+                  <span className="data-source-bulk-toolbar__stat-label">内置</span>
+                  <strong>{builtInSources.length}</strong>
+                </div>
+                <div className="data-source-bulk-toolbar__stat-pill">
+                  <span className="data-source-bulk-toolbar__stat-label">已启用</span>
+                  <strong>{activeBuiltInCount}</strong>
+                </div>
+                <div className="data-source-bulk-toolbar__stat-pill">
+                  <span className="data-source-bulk-toolbar__stat-label">执行中</span>
+                  <strong>{bulkProgressBatch ? bulkBatchRunningCount : runningBuiltInCount}</strong>
+                </div>
+                {bulkProgressBatch ? (
+                  <div className="data-source-bulk-toolbar__stat-pill data-source-bulk-toolbar__stat-pill--success">
+                    <span className="data-source-bulk-toolbar__stat-label">成功</span>
+                    <strong>{bulkBatchSuccessCount}/{bulkProgressBatch.sourceIds.length}</strong>
+                  </div>
+                ) : null}
+                {bulkProgressBatch ? (
+                  <div className="data-source-bulk-toolbar__stat-pill data-source-bulk-toolbar__stat-pill--danger">
+                    <span className="data-source-bulk-toolbar__stat-label">失败</span>
+                    <strong>{bulkBatchFailedCount}</strong>
+                  </div>
+                ) : null}
              </div>
            </div>
            <Space size={12} align="center">
@@ -796,6 +942,7 @@ function DataSources() {

  return (
    <AppLayout>
+      {contextHolder}
      <div className="page-shell">
        <div className="page-shell__header">
          <h2 style={{ margin: 0 }}>数据源管理</h2>
--- a/frontend/src/pages/Settings/Settings.tsx
+++ b/frontend/src/pages/Settings/Settings.tsx
@@ -7,7 +7,6 @@ import {
  InputNumber,
  message,
  Select,
-  Space,
  Switch,
  Table,
  Tabs,