feat:抢单大厅数据排列优化
This commit is contained in:
@@ -62,10 +62,13 @@ export async function createRealtimeEvent(input: {
|
||||
export async function listRealtimeEventsAfter(eventId: number, limit = 500) {
|
||||
const result = await query<RealtimeEventRow>(
|
||||
`
|
||||
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))],
|
||||
)
|
||||
|
||||
@@ -85,6 +85,24 @@ let nextSubscriberId = 1
|
||||
let heartbeatTimer: NodeJS.Timeout | null = null
|
||||
let publishQueue: Promise<void> = 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
|
||||
|
||||
@@ -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,让列表整体重拉(接口才是权威数据源)',
|
||||
)
|
||||
})
|
||||
|
||||
@@ -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
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
Reference in New Issue
Block a user