diff --git a/apps/backend/src/repositories/realtime-event-repo.ts b/apps/backend/src/repositories/realtime-event-repo.ts index 31b40d11..bcc2c6f8 100644 --- a/apps/backend/src/repositories/realtime-event-repo.ts +++ b/apps/backend/src/repositories/realtime-event-repo.ts @@ -62,10 +62,13 @@ export async function createRealtimeEvent(input: { export async function listRealtimeEventsAfter(eventId: number, limit = 500) { const result = await query( ` - SELECT * FROM realtime_events - WHERE id > $1 + SELECT * FROM ( + SELECT * FROM realtime_events + WHERE id > $1 + ORDER BY id DESC + LIMIT $2 + ) replay_window ORDER BY id ASC - LIMIT $2 `, [Math.max(0, Number(eventId) || 0), Math.min(500, Math.max(1, Number(limit) || 500))], ) diff --git a/apps/backend/src/services/realtime/realtime-event-service.ts b/apps/backend/src/services/realtime/realtime-event-service.ts index c99eee1f..b2572084 100644 --- a/apps/backend/src/services/realtime/realtime-event-service.ts +++ b/apps/backend/src/services/realtime/realtime-event-service.ts @@ -85,6 +85,24 @@ let nextSubscriberId = 1 let heartbeatTimer: NodeJS.Timeout | null = null let publishQueue: Promise = Promise.resolve() +/** + * 单次补发的事件上限,与 realtime_events 表保留的条数一致 + * (见 realtime-event-repo 的清理策略:只保留最近 500 条)。 + */ +export const REALTIME_REPLAY_EVENT_LIMIT = 500 + +/** + * 是否应该补发历史事件。 + * + * 只有断线重连带上了 Last-Event-ID(游标 > 0)才补发。新连接(游标 = 0)**不补发**: + * 补发的每一条都携带「事件发生那一刻」的工单快照,直接打进刚由列表接口加载好的缓存, + * 会把已涨价/已拼满的订单覆盖回旧值(表现为「进大厅显示旧单价旧进度,点一次刷新才对」)。 + */ +export function shouldReplayRealtimeEvents(lastEventId: number): boolean { + const normalized = Number(lastEventId) + return Number.isFinite(normalized) && normalized > 0 +} + /** 建立后台实时事件流,事件快照用于前端定点更新缓存。 */ export function openAdminRealtimeStream(response: Response, lastEventId = 0) { return openRealtimeStream(response, null, lastEventId) @@ -532,7 +550,10 @@ function openRealtimeStream(response: Response, workerId: number | null, lastEve connectedAt: new Date().toISOString(), lastEventId: Math.max(0, Number(lastEventId) || 0), }) - void replayEvents(subscriber, Math.max(0, Number(lastEventId) || 0)) + // 新连接不补发历史事件,只有断线重连(带游标)才回放缺失的增量。 + if (shouldReplayRealtimeEvents(lastEventId)) { + void replayEvents(subscriber, Number(lastEventId)) + } ensureHeartbeat() response.on('close', () => { @@ -542,7 +563,16 @@ function openRealtimeStream(response: Response, workerId: number | null, lastEve } async function replayEvents(subscriber: RealtimeSubscriber, lastEventId: number) { - const events = await listRealtimeEventsAfter(lastEventId) + const events = await listRealtimeEventsAfter(lastEventId, REALTIME_REPLAY_EVENT_LIMIT) + // 补发窗口被拿满 = 更早的事件已经从事件表里滚掉(断线太久)。 + // 这时逐条套用旧快照只会把列表写回旧值,直接让客户端整体重拉一次。 + if (events.length >= REALTIME_REPLAY_EVENT_LIMIT) { + writeEvent(subscriber, 'resync', { + headEventId: Number(events[events.length - 1]?.id || 0), + reason: 'replay_window_overflow', + }) + return + } for (const row of events) { const event = mapStoredEvent(row) if (event.eventId <= lastEventId || !isEventVisibleToSubscriber(event, subscriber)) continue diff --git a/apps/backend/src/services/realtime/realtime-replay.test.ts b/apps/backend/src/services/realtime/realtime-replay.test.ts index 7df17f5c..de60e5ee 100644 --- a/apps/backend/src/services/realtime/realtime-replay.test.ts +++ b/apps/backend/src/services/realtime/realtime-replay.test.ts @@ -1,4 +1,5 @@ import assert from 'node:assert/strict' +import fs from 'node:fs' import test from 'node:test' import { shouldReplayRealtimeEvents } from './realtime-event-service.js' @@ -20,3 +21,36 @@ test('断线重连(带 Last-Event-ID)才补发事件', () => { assert.equal(shouldReplayRealtimeEvents(1), true) assert.equal(shouldReplayRealtimeEvents(3993659), true) }) + +test('补发窗口取最新的一段,而不是最旧的一段', () => { + const source = fs.readFileSync( + new URL('../../repositories/realtime-event-repo.ts', import.meta.url), + 'utf8', + ) + const block = source.slice( + source.indexOf('export async function listRealtimeEventsAfter('), + source.indexOf('\n}', source.indexOf('export async function listRealtimeEventsAfter(')), + ) + assert.ok(block.length > 0, 'listRealtimeEventsAfter 片段必须存在') + assert.ok( + /ORDER BY id DESC\s+LIMIT \$2\s+\)\s*replay_window\s+ORDER BY id ASC/s.test(block), + '补发必须「内层 id DESC 取最新 N 条、外层 id ASC 回放」;否则断线久了窗口会停在旧事件上', + ) +}) + +test('补发窗口溢出时改为让客户端整体重拉', () => { + const source = fs.readFileSync(new URL('./realtime-event-service.ts', import.meta.url), 'utf8') + const replayBlock = source.slice( + source.indexOf('async function replayEvents('), + source.indexOf('function mapStoredEvent('), + ) + assert.ok(replayBlock.length > 0, 'replayEvents 片段必须存在') + assert.ok( + replayBlock.includes('events.length >= REALTIME_REPLAY_EVENT_LIMIT'), + '补发窗口被拿满时必须判定为溢出,不能继续逐条套用旧快照', + ) + assert.ok( + replayBlock.includes("'resync'"), + '溢出时要给客户端发 resync,让列表整体重拉(接口才是权威数据源)', + ) +}) diff --git a/apps/frontend/src/hooks/useRealtimeEvents.ts b/apps/frontend/src/hooks/useRealtimeEvents.ts index f22d5900..ea3c88cc 100644 --- a/apps/frontend/src/hooks/useRealtimeEvents.ts +++ b/apps/frontend/src/hooks/useRealtimeEvents.ts @@ -21,6 +21,13 @@ type UseRealtimeEventsOptions = { token: string onEvent: (event: RealtimeEvent) => void onConnected?: () => void + /** + * 服务端告知「补发窗口溢出,历史事件已不连续」。 + * + * 此时不能再依赖逐条快照补丁(会把列表写回旧值),调用方应整体失效对应查询, + * 让列表接口重新给出权威数据。 + */ + onResync?: (info: { headEventId: number; reason: string }) => void /** 鉴权失效(401/403)时触发:连接停止重连,调用方应清理会话并跳转登录页。 */ onAuthExpired?: () => void } @@ -33,13 +40,16 @@ export function useRealtimeEvents({ token, onEvent, onConnected, + onResync, onAuthExpired, }: UseRealtimeEventsOptions) { const onEventRef = useRef(onEvent) const onConnectedRef = useRef(onConnected) + const onResyncRef = useRef(onResync) const onAuthExpiredRef = useRef(onAuthExpired) onEventRef.current = onEvent onConnectedRef.current = onConnected + onResyncRef.current = onResync onAuthExpiredRef.current = onAuthExpired useEffect(() => { @@ -79,6 +89,15 @@ export function useRealtimeEvents({ } onConnectedRef.current?.() await readSseStream(response.body, (eventName, data) => { + // 补发窗口溢出:历史事件不连续,不能逐条复位快照,交给调用方整体重拉。 + if (eventName === 'resync') { + const notice = normalizeResyncNotice(data) + if (notice) { + lastEventId = Math.max(lastEventId, notice.headEventId) + onResyncRef.current?.(notice) + } + return + } if (eventName !== 'realtime') return const event = normalizeRealtimeEvent(data) if (event) { @@ -197,3 +216,16 @@ function normalizeRealtimeEvent(value: string): RealtimeEvent | null { return null } } + +/** 服务端 resync 通知:补发窗口溢出,客户端应把游标推进到 headEventId 并整体重拉。 */ +function normalizeResyncNotice(value: string): { headEventId: number; reason: string } | null { + try { + const parsed = JSON.parse(value) as { headEventId?: unknown; reason?: unknown } + return { + headEventId: Math.max(0, Number(parsed.headEventId) || 0), + reason: String(parsed.reason || ''), + } + } catch { + return null + } +} diff --git a/apps/frontend/src/layouts/AdminLayout.tsx b/apps/frontend/src/layouts/AdminLayout.tsx index babc4e64..edb35e1f 100644 --- a/apps/frontend/src/layouts/AdminLayout.tsx +++ b/apps/frontend/src/layouts/AdminLayout.tsx @@ -119,6 +119,16 @@ export default function AdminLayout() { url: '/api/v1/admin/realtime', token: role === 'finance' ? '' : getAdminToken(), onEvent: (event) => applyAdminRealtimeEvent(event, queryClient), + // 补发窗口溢出(断线太久):历史事件已不连续,整体失效让接口重新给权威数据。 + onResync: () => { + void queryClient.invalidateQueries({ queryKey: ['admin-worker-platform-orders'] }) + void queryClient.invalidateQueries({ queryKey: ['admin-worker-platform-after-sales-cases'] }) + void queryClient.invalidateQueries({ queryKey: ['admin-worker-platform-finance-requests'] }) + void queryClient.invalidateQueries({ queryKey: ['admin-notifications'] }) + void queryClient.invalidateQueries({ queryKey: ['admin-tasks'] }) + void queryClient.invalidateQueries({ queryKey: ['admin-orders'] }) + void queryClient.invalidateQueries({ queryKey: ['admin-dashboard-summary'] }) + }, // 会话过期时清登录态回登录页,避免 SSE 带着失效 token 无限重连刷 401。 onAuthExpired: () => { clearAdminSession() diff --git a/apps/frontend/src/layouts/WorkerLayout.tsx b/apps/frontend/src/layouts/WorkerLayout.tsx index 26c34b48..d697d8c0 100644 --- a/apps/frontend/src/layouts/WorkerLayout.tsx +++ b/apps/frontend/src/layouts/WorkerLayout.tsx @@ -138,6 +138,16 @@ export default function WorkerLayout() { url: '/api/v1/worker/realtime', token: getWorkerToken(), onEvent: (event) => applyWorkerRealtimeEvent(event, queryClient), + // 补发窗口溢出(断线太久):历史事件已不连续,逐条套旧快照会把列表写回旧值, + // 这里整体失效,让列表接口重新给权威数据。 + onResync: () => { + void queryClient.invalidateQueries({ queryKey: ['worker-hall-orders'] }) + void queryClient.invalidateQueries({ queryKey: ['worker-my-orders'] }) + void queryClient.invalidateQueries({ queryKey: ['worker-my-orders-overview'] }) + void queryClient.invalidateQueries({ queryKey: ['worker-after-sales-cases'] }) + void queryClient.invalidateQueries({ queryKey: ['worker-finance-requests'] }) + void queryClient.invalidateQueries({ queryKey: ['worker-profile'] }) + }, // 会话过期时清登录态回登录页,避免 SSE 带着失效 token 无限重连刷 401。 onAuthExpired: () => { expireWorkerSession()