diff --git a/backend/internal/handlers/stock.go b/backend/internal/handlers/stock.go index fb90c11..b941b52 100644 --- a/backend/internal/handlers/stock.go +++ b/backend/internal/handlers/stock.go @@ -147,7 +147,7 @@ func (h *StockHandler) SyncStatus(c *gin.Context) { } // TriggerSync 手动触发同步(管理员)。 -// 系统管理员点击时同步近半年历史数据,其他管理员只同步当天。 +// 系统管理员点击时异步同步近半年历史数据,其他管理员只同步当天。 func (h *StockHandler) TriggerSync(c *gin.Context) { user, ok := middleware.GetCurrentUser(c) if !ok { @@ -155,18 +155,18 @@ func (h *StockHandler) TriggerSync(c *gin.Context) { return } - var count int + var job *models.StockSyncJob var err error if user.Role == models.RoleSystemAdmin { - count, err = h.Svc.SyncHistory(c.Request.Context(), "manual", 6) + job, err = h.Svc.StartSyncHistory(c.Request.Context(), "manual", 6) } else { - count, err = h.Svc.Sync(c.Request.Context(), "manual") + job, err = h.Svc.StartSync(c.Request.Context(), "manual") } if err != nil { c.JSON(http.StatusInternalServerError, gin.H{"success": false, "error": err.Error()}) return } - c.JSON(http.StatusOK, gin.H{"success": true, "records_count": count}) + c.JSON(http.StatusOK, gin.H{"success": true, "job": job}) } type distributionRow struct { diff --git a/backend/internal/services/stock_sync.go b/backend/internal/services/stock_sync.go index b522922..ef62cc6 100644 --- a/backend/internal/services/stock_sync.go +++ b/backend/internal/services/stock_sync.go @@ -52,10 +52,31 @@ func (s *StockSyncService) SyncHistory(ctx context.Context, triggerBy string, mo }) } -func (s *StockSyncService) runSyncJob(ctx context.Context, triggerBy string, fn func() (int, error)) (int, error) { - s.mu.Lock() - defer s.mu.Unlock() +// StartSync 异步启动一次盘后同步,返回已创建的 job。 +func (s *StockSyncService) StartSync(ctx context.Context, triggerBy string) (*models.StockSyncJob, error) { + job, err := s.createJob(ctx, triggerBy) + if err != nil { + return nil, err + } + go s.runSyncJobByID(context.Background(), job.ID, func(ctx context.Context) (int, error) { + return s.doSync(ctx) + }) + return job, nil +} +// StartSyncHistory 异步启动近 N 个月历史同步,返回已创建的 job。 +func (s *StockSyncService) StartSyncHistory(ctx context.Context, triggerBy string, months int) (*models.StockSyncJob, error) { + job, err := s.createJob(ctx, triggerBy) + if err != nil { + return nil, err + } + go s.runSyncJobByID(context.Background(), job.ID, func(ctx context.Context) (int, error) { + return s.doSyncHistory(ctx, months) + }) + return job, nil +} + +func (s *StockSyncService) createJob(ctx context.Context, triggerBy string) (*models.StockSyncJob, error) { job := models.StockSyncJob{ JobDate: today(), Status: "running", @@ -63,25 +84,46 @@ func (s *StockSyncService) runSyncJob(ctx context.Context, triggerBy string, fn TriggerBy: triggerBy, } if err := s.db.WithContext(ctx).Create(&job).Error; err != nil { - return 0, fmt.Errorf("create sync job: %w", err) + return nil, fmt.Errorf("create sync job: %w", err) } + return &job, nil +} +func (s *StockSyncService) runSyncJob(ctx context.Context, triggerBy string, fn func() (int, error)) (int, error) { + job, err := s.createJob(ctx, triggerBy) + if err != nil { + return 0, err + } recordsCount, err := fn() + s.finishJob(ctx, job.ID, recordsCount, err) + return recordsCount, err +} +func (s *StockSyncService) runSyncJobByID(ctx context.Context, jobID string, fn func(context.Context) (int, error)) { + s.mu.Lock() + defer s.mu.Unlock() + + recordsCount, err := fn(ctx) + s.finishJob(ctx, jobID, recordsCount, err) +} + +func (s *StockSyncService) finishJob(ctx context.Context, jobID string, recordsCount int, jobErr error) { now := time.Now().UTC() updates := map[string]any{ "records_count": recordsCount, "finished_at": &now, } - if err != nil { + if jobErr != nil { updates["status"] = "failed" - updates["error_message"] = err.Error() + updates["error_message"] = jobErr.Error() + log.Printf("[stock sync] job %s failed: %v", jobID, jobErr) } else { updates["status"] = "success" + log.Printf("[stock sync] job %s success, records=%d", jobID, recordsCount) + } + if err := s.db.WithContext(ctx).Model(&models.StockSyncJob{}).Where("id = ?", jobID).Updates(updates).Error; err != nil { + log.Printf("[stock sync] update job %s failed: %v", jobID, err) } - s.db.WithContext(ctx).Model(&job).Updates(updates) - - return recordsCount, err } func (s *StockSyncService) doSync(ctx context.Context) (int, error) { diff --git a/frontend/src/components/StockOverview.tsx b/frontend/src/components/StockOverview.tsx index c1dde35..2672106 100644 --- a/frontend/src/components/StockOverview.tsx +++ b/frontend/src/components/StockOverview.tsx @@ -1,4 +1,4 @@ -import { useEffect, useState } from 'react' +import { useEffect, useRef, useState } from 'react' import { BarChart3, TrendingUp, @@ -24,6 +24,7 @@ export function StockOverview({ role }: StockOverviewProps) { const [loading, setLoading] = useState(true) const [syncing, setSyncing] = useState(false) const [error, setError] = useState('') + const pollRef = useRef(null) const canSync = canManageUsers(role) @@ -43,20 +44,52 @@ export function StockOverview({ role }: StockOverviewProps) { useEffect(() => { fetchAll() + return () => { + if (pollRef.current) { + window.clearInterval(pollRef.current) + } + } }, []) + const startPolling = () => { + if (pollRef.current) { + window.clearInterval(pollRef.current) + } + pollRef.current = window.setInterval(async () => { + try { + const j = await api.stockSyncStatus() + setJob(j) + if (!j || j.status !== 'running') { + if (pollRef.current) { + window.clearInterval(pollRef.current) + pollRef.current = null + } + setSyncing(false) + await fetchAll() + if (j?.status === 'failed') { + setError(j.error_message || '同步失败') + } + } + } catch (err: any) { + if (pollRef.current) { + window.clearInterval(pollRef.current) + pollRef.current = null + } + setSyncing(false) + setError(err?.message || '轮询同步状态失败') + } + }, 2000) + } + const handleSync = async () => { setSyncing(true) setError('') try { - const res = await api.triggerStockSync() - await fetchAll() - // eslint-disable-next-line no-console - console.log('synced', res.records_count, 'records') + await api.triggerStockSync() + startPolling() } catch (err: any) { - setError(err?.message || '同步失败') - } finally { setSyncing(false) + setError(err?.message || '同步失败') } } diff --git a/frontend/src/lib/api.ts b/frontend/src/lib/api.ts index 11bc79e..99cee31 100644 --- a/frontend/src/lib/api.ts +++ b/frontend/src/lib/api.ts @@ -58,7 +58,7 @@ export interface SyncJob { records_count: number started_at: string finished_at?: string - error_message: string + error_message?: string trigger_by: string } @@ -147,5 +147,5 @@ export const api = { stockSyncStatus: () => request('/api/stocks/sync/status'), triggerStockSync: () => - request<{ records_count: number }>('/api/admin/stocks/sync', { method: 'POST' }), + request<{ job: SyncJob }>('/api/admin/stocks/sync', { method: 'POST' }), }