656 lines
23 KiB
TypeScript
656 lines
23 KiB
TypeScript
import type { Response } from 'express'
|
|
|
|
import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js'
|
|
|
|
import {
|
|
getLatestMaterialEventAtByOrderIds,
|
|
getPendingCancelRequestOrderIds,
|
|
getPendingFeedbackOrderIds,
|
|
getWorkerFinanceRequestById,
|
|
getWorkerUserById,
|
|
getWorkOrderById,
|
|
listWorkerTimeoutEventsByOrderIds,
|
|
listWorkerWorkOrderNotes,
|
|
listWorkerWorkOrderViews,
|
|
listWorkOrderShares,
|
|
sumPendingUnfreezeByOrderIds,
|
|
} from '../../repositories/worker-platform/index.js'
|
|
import { getOrderById } from '../../repositories/order-repo.js'
|
|
import { listTasksByOrderId } from '../../repositories/task-repo.js'
|
|
import {
|
|
createRealtimeEvent,
|
|
listRealtimeEventsAfter,
|
|
type RealtimeEventRow,
|
|
} from '../../repositories/realtime-event-repo.js'
|
|
import { getAdminDashboardSummary } from '../admin/admin-dashboard-service.js'
|
|
import { mapAdminOrderListItem } from '../admin/admin-order-read-helpers.js'
|
|
import { mapAdminTaskListItem } from '../admin/admin-task-read-helpers.js'
|
|
import {
|
|
isWorkOrderSharingFilled,
|
|
isWorkerOrderVipEvidencePending,
|
|
mapFinanceRequest,
|
|
mapWorkOrderAdmin,
|
|
mapWorkOrderForWorker,
|
|
mapWorkerUser,
|
|
pickWorkerRepresentativeShare,
|
|
resolveWorkerOrderStatusTabs,
|
|
resolveWorkerPermissions,
|
|
} from '../worker-platform/mappers.js'
|
|
import { invalidateWorkerMyOrderCaches } from '../worker-platform/worker-my-order-cache.js'
|
|
|
|
export type RealtimeEventType =
|
|
| 'admin_notification.changed'
|
|
| 'order.changed'
|
|
| 'work_order.changed'
|
|
| 'worker_finance_request.changed'
|
|
| 'worker_wallet.changed'
|
|
|
|
export type RealtimeEvent = {
|
|
eventId: number
|
|
version: number
|
|
type: RealtimeEventType
|
|
entityId: number
|
|
scopes: string[]
|
|
operation: 'upsert' | 'remove' | 'refresh'
|
|
data?: unknown
|
|
occurredAt: string
|
|
}
|
|
|
|
type RealtimeEventInput = {
|
|
type: RealtimeEventType
|
|
entityId?: number
|
|
scopes?: string[]
|
|
operation?: RealtimeEvent['operation']
|
|
data?: unknown
|
|
admin?: boolean
|
|
workerIds?: number[]
|
|
broadcastWorkers?: boolean
|
|
}
|
|
|
|
type RealtimeSubscriber = {
|
|
id: number
|
|
response: Response
|
|
workerId: number | null
|
|
}
|
|
|
|
type StoredRealtimeEvent = RealtimeEvent & {
|
|
adminAudience: boolean
|
|
workerIds: number[]
|
|
broadcastWorkers: boolean
|
|
}
|
|
|
|
const adminSubscribers = new Map<number, RealtimeSubscriber>()
|
|
const workerSubscribers = new Map<number, RealtimeSubscriber>()
|
|
let nextSubscriberId = 1
|
|
let heartbeatTimer: NodeJS.Timeout | null = null
|
|
let publishQueue: Promise<void> = Promise.resolve()
|
|
|
|
/** 建立后台实时事件流，事件快照用于前端定点更新缓存。 */
|
|
export function openAdminRealtimeStream(response: Response, lastEventId = 0) {
|
|
return openRealtimeStream(response, null, lastEventId)
|
|
}
|
|
|
|
/** 建立打手实时事件流，只向该打手推送其可见的订单和资金变化。 */
|
|
export function openWorkerRealtimeStream(response: Response, workerId: number, lastEventId = 0) {
|
|
return openRealtimeStream(response, Number(workerId), lastEventId)
|
|
}
|
|
|
|
/** 发布带版本号和数据快照的实时事件；数据库仍是最终权威数据源。 */
|
|
export function publishRealtimeEvent(input: RealtimeEventInput) {
|
|
const workerIds = [
|
|
...new Set(
|
|
(input.workerIds || [])
|
|
.map((item) => Number(item))
|
|
.filter((item) => Number.isInteger(item) && item > 0),
|
|
),
|
|
]
|
|
const event = {
|
|
type: input.type,
|
|
entityId: Math.max(0, Number(input.entityId) || 0),
|
|
scopes: [
|
|
...new Set((input.scopes || []).map((item) => String(item || '').trim()).filter(Boolean)),
|
|
],
|
|
operation: input.operation || 'refresh',
|
|
data: input.data,
|
|
adminAudience: input.admin === true,
|
|
workerIds,
|
|
broadcastWorkers: input.broadcastWorkers === true,
|
|
occurredAt: new Date().toISOString(),
|
|
}
|
|
// 按提交顺序串行写入，保证 Last-Event-ID 与推送顺序一致。
|
|
publishQueue = publishQueue.then(() => persistAndBroadcast(event)).catch(() => undefined)
|
|
}
|
|
|
|
async function persistAndBroadcast(input: {
|
|
type: RealtimeEventType
|
|
entityId: number
|
|
scopes: string[]
|
|
operation: RealtimeEvent['operation']
|
|
data?: unknown
|
|
adminAudience: boolean
|
|
workerIds: number[]
|
|
broadcastWorkers: boolean
|
|
occurredAt: string
|
|
}) {
|
|
const row = await createRealtimeEvent({
|
|
eventType: input.type,
|
|
entityId: input.entityId,
|
|
scopes: input.scopes,
|
|
operation: input.operation,
|
|
data: input.data,
|
|
adminAudience: input.adminAudience,
|
|
workerIds: input.workerIds,
|
|
broadcastWorkers: input.broadcastWorkers,
|
|
occurredAt: input.occurredAt,
|
|
})
|
|
const storedEvent = mapStoredEvent(row)
|
|
if (storedEvent.adminAudience) {
|
|
broadcast(adminSubscribers.values(), storedEvent)
|
|
}
|
|
if (storedEvent.broadcastWorkers) {
|
|
broadcast(workerSubscribers.values(), storedEvent)
|
|
return
|
|
}
|
|
if (storedEvent.workerIds.length > 0) {
|
|
broadcast(
|
|
[...workerSubscribers.values()].filter((subscriber) =>
|
|
storedEvent.workerIds.includes(Number(subscriber.workerId)),
|
|
),
|
|
storedEvent,
|
|
)
|
|
}
|
|
}
|
|
|
|
/** 工单变更统一通知后台；大厅与打手本人按可见范围分别接收。 */
|
|
export function publishWorkOrderRealtimeChange(input: {
|
|
workOrderId: number
|
|
workerIds?: number[]
|
|
hallChanged?: boolean
|
|
}) {
|
|
const workOrderId = Math.max(0, Number(input.workOrderId) || 0)
|
|
void publishWorkOrderSnapshots({ ...input, workOrderId }).catch(() => {
|
|
publishRealtimeEvent({
|
|
type: 'work_order.changed',
|
|
entityId: workOrderId,
|
|
scopes: ['admin_work_orders'],
|
|
operation: 'refresh',
|
|
admin: true,
|
|
})
|
|
})
|
|
}
|
|
|
|
/** 大厅全局配置变化时刷新所有打手的候选集合。 */
|
|
export function publishWorkerHallRealtimeRefresh() {
|
|
publishRealtimeEvent({
|
|
type: 'work_order.changed',
|
|
scopes: ['worker_hall'],
|
|
operation: 'refresh',
|
|
broadcastWorkers: true,
|
|
})
|
|
}
|
|
|
|
/** 资金申请变更会同步后台资金页与申请所属打手的钱包页面。 */
|
|
export function publishWorkerFinanceRealtimeChange(input: {
|
|
requestId: number
|
|
workerId: number
|
|
walletChanged?: boolean
|
|
}) {
|
|
void publishWorkerFinanceSnapshots(input).catch(() => {
|
|
publishRealtimeEvent({
|
|
type: 'worker_finance_request.changed',
|
|
entityId: input.requestId,
|
|
scopes: ['admin_worker_finance'],
|
|
operation: 'refresh',
|
|
admin: true,
|
|
})
|
|
})
|
|
}
|
|
|
|
/** 钱包事件携带当前钱包快照，避免客户端为余额变化重新请求个人资料。 */
|
|
export function publishWorkerWalletRealtimeChange(workerId: number) {
|
|
const normalizedWorkerId = Math.max(0, Number(workerId) || 0)
|
|
void getWorkerUserById(normalizedWorkerId)
|
|
.then((worker) => {
|
|
publishRealtimeEvent({
|
|
type: 'worker_wallet.changed',
|
|
entityId: normalizedWorkerId,
|
|
scopes: ['worker_wallet'],
|
|
operation: worker ? 'upsert' : 'refresh',
|
|
...(worker ? { data: { wallet: mapWorkerUser(worker).wallet } } : {}),
|
|
workerIds: [normalizedWorkerId],
|
|
})
|
|
})
|
|
.catch(() => undefined)
|
|
}
|
|
|
|
/** 订单变更推送后台列表行快照，前端无需重新读取整页订单。 */
|
|
export function publishAdminOrderRealtimeChange(orderId: number) {
|
|
const normalizedOrderId = Math.max(0, Number(orderId) || 0)
|
|
void getOrderById(normalizedOrderId)
|
|
.then(async (order) => {
|
|
const [dashboard, tasks] = await Promise.all([
|
|
getAdminDashboardSummary(),
|
|
order ? listTasksByOrderId(normalizedOrderId) : [],
|
|
])
|
|
const data = {
|
|
order: order ? await mapAdminOrderListItem(order) : { orderId: normalizedOrderId },
|
|
dashboard,
|
|
tasks: tasks.map((task) => mapAdminTaskListItem(task)),
|
|
}
|
|
publishRealtimeEvent({
|
|
type: 'order.changed',
|
|
entityId: normalizedOrderId,
|
|
scopes: ['admin_orders'],
|
|
operation: order ? 'upsert' : 'remove',
|
|
data,
|
|
admin: true,
|
|
})
|
|
})
|
|
.catch(() => undefined)
|
|
}
|
|
|
|
async function publishWorkOrderSnapshots(input: {
|
|
workOrderId: number
|
|
workerIds?: number[]
|
|
hallChanged?: boolean
|
|
}) {
|
|
const workOrderId = Math.max(0, Number(input.workOrderId) || 0)
|
|
// 调用方显式给出的受影响打手必须同步失效：下面的快照生成要先查库，
|
|
// 若等都查完再失效，紧接着到达的「我的订单」请求会命中旧缓存（30s TTL）。
|
|
invalidateWorkerMyOrderCaches(input.workerIds || [])
|
|
const workOrder = await getWorkOrderById(workOrderId)
|
|
if (!workOrder) {
|
|
// 工单已不存在（删除）：调用方传入的打手集合仍需失效缓存，
|
|
// 否则他们的「我的订单」会继续展示已删除的单。
|
|
invalidateWorkerMyOrderCaches(input.workerIds || [])
|
|
publishRealtimeEvent({
|
|
type: 'work_order.changed',
|
|
entityId: workOrderId,
|
|
scopes: ['admin_work_orders'],
|
|
operation: 'remove',
|
|
data: { workOrderId },
|
|
admin: true,
|
|
})
|
|
return
|
|
}
|
|
|
|
const pendingUnfreezeByOrderId = await sumPendingUnfreezeByOrderIds([workOrderId])
|
|
const shares = await listWorkOrderShares(workOrderId)
|
|
|
|
// 事件驱动缓存失效：这里已经拿到的工单与拼单快照恰好给出完整的受影响打手集合
|
|
// （当前指派 + 历史保留 + 全部拼单人 + 调用方显式指定的）。
|
|
// 先失效再广播，保证打手端随后的拉取一定看到新数据；这也是「我的订单」缓存
|
|
// 能放心用 30s 长 TTL 的前提（见 worker-my-order-cache.ts）。
|
|
invalidateWorkerMyOrderCaches([
|
|
...(input.workerIds || []),
|
|
workOrder.assigned_worker_id,
|
|
workOrder.last_assigned_worker_id,
|
|
...shares.map((share) => share.worker_id),
|
|
])
|
|
|
|
const joinedQuantity = shares
|
|
.filter((share) => share.status !== 'cancelled')
|
|
.reduce((sum, share) => sum + Number(share.quantity || 0), 0)
|
|
const pendingSubmissionCount = shares.filter((share) => share.status === 'joined').length
|
|
const adminSnapshot = {
|
|
...mapWorkOrderAdmin(workOrder, shares),
|
|
pendingUnfreezeAmount: pendingUnfreezeByOrderId.get(workOrderId) || 0,
|
|
sharingProgress: {
|
|
joinedQuantity,
|
|
pendingSubmissionCount,
|
|
remainingQuantity: Math.max(
|
|
0,
|
|
Number(workOrder.sharing_total_quantity || 1) - joinedQuantity,
|
|
),
|
|
filled: isWorkOrderSharingFilled(
|
|
workOrder.sharing_enabled === true,
|
|
Number(workOrder.sharing_total_quantity || 1),
|
|
joinedQuantity,
|
|
),
|
|
},
|
|
}
|
|
|
|
publishRealtimeEvent({
|
|
type: 'work_order.changed',
|
|
entityId: workOrderId,
|
|
scopes: ['admin_work_orders'],
|
|
operation: 'upsert',
|
|
data: adminSnapshot,
|
|
admin: true,
|
|
})
|
|
|
|
if (input.hallChanged) {
|
|
// 携带快照的精确事件让客户端原地增删缓存行，避免整页刷新导致抢单误点。
|
|
const hallVisible =
|
|
workOrder.status === WORK_ORDER_STATUS.OPEN &&
|
|
!isWorkOrderSharingFilled(
|
|
workOrder.sharing_enabled === true,
|
|
Number(workOrder.sharing_total_quantity || 1),
|
|
joinedQuantity,
|
|
)
|
|
publishRealtimeEvent({
|
|
type: 'work_order.changed',
|
|
entityId: workOrderId,
|
|
scopes: ['worker_hall'],
|
|
operation: hallVisible ? 'upsert' : 'remove',
|
|
data: hallVisible
|
|
? {
|
|
...mapWorkOrderForWorker(workOrder, { depositFreeAmount: 0 }),
|
|
requiredDepositAmount: Number(workOrder.required_deposit_amount || 0),
|
|
// 广播无法按打手等级过滤可见延迟，携带入厅时间由客户端按各自延迟推迟展示。
|
|
hallQueuedAt: workOrder.hall_queued_at || null,
|
|
sharingProgress: { joinedQuantity, pendingSubmissionCount },
|
|
}
|
|
: { workOrderId },
|
|
broadcastWorkers: true,
|
|
})
|
|
}
|
|
|
|
// 资料更新时间只与工单有关，提到循环外只查一次（参与人数多时省掉 N 次查询）。
|
|
const materialEventAt =
|
|
(await getLatestMaterialEventAtByOrderIds([workOrderId])).get(workOrderId) || ''
|
|
|
|
for (const workerId of new Set(
|
|
(input.workerIds || []).map((item) => Number(item)).filter(Boolean),
|
|
)) {
|
|
const worker = await getWorkerUserById(workerId)
|
|
if (!worker) continue
|
|
|
|
// 同一打手在同一工单可能有多份份额（多次拼单），必须按「全部份额」还原这一行：
|
|
// 旧实现用 shares.find(...) 只取最早一条，多份额时会把可见性、个人状态、金额都算错，
|
|
// 还会让实时补丁把行从错误的页签删掉（打手端偶发「页签里没有数据」）。
|
|
const isAssignedWorker = Number(workOrder.assigned_worker_id || 0) === workerId
|
|
const workerShares = shares.filter((item) => Number(item.worker_id) === workerId)
|
|
const activeWorkerShares = workerShares.filter((item) => item.status !== 'cancelled')
|
|
const representativeShare = pickWorkerRepresentativeShare(
|
|
activeWorkerShares.length > 0 ? activeWorkerShares : workerShares,
|
|
)
|
|
const cancelledShare = workerShares
|
|
.filter((item) => item.status === 'cancelled')
|
|
.sort((left, right) => Number(right.id) - Number(left.id))[0]
|
|
const isOwnerCurrently = isAssignedWorker || activeWorkerShares.length > 0
|
|
const workerOrderVisible = isOwnerCurrently
|
|
|
|
// 只有「已不持有该单」时才需要判断超时归属，避免给正常在办的单多打一次查询。
|
|
const needsTimeoutCheck = !isOwnerCurrently
|
|
const [notes, views, pendingCancelOrderIds, pendingFeedbackOrderIds, timeoutEventsByOrderId] =
|
|
await Promise.all([
|
|
listWorkerWorkOrderNotes(workerId, [workOrderId]),
|
|
listWorkerWorkOrderViews(workerId, [workOrderId]),
|
|
getPendingCancelRequestOrderIds(workerId, [workOrderId]),
|
|
getPendingFeedbackOrderIds(workerId, [workOrderId]),
|
|
needsTimeoutCheck
|
|
? listWorkerTimeoutEventsByOrderIds(workerId, [workOrderId])
|
|
: Promise.resolve(new Map<number, { createdAt: string }>()),
|
|
])
|
|
const note = notes.find((item) => Number(item.work_order_id) === workOrderId)?.note || ''
|
|
const view = views.find((item) => Number(item.work_order_id) === workOrderId)
|
|
const viewedAt = view?.viewed_at || null
|
|
const timeoutEvent = timeoutEventsByOrderId.get(workOrderId)
|
|
const isTimedOutForMe =
|
|
!isOwnerCurrently && Boolean(timeoutEvent || (cancelledShare && workOrder.sharing_enabled))
|
|
|
|
const timeoutNotice = isTimedOutForMe
|
|
? {
|
|
isTimedOut: true,
|
|
timedOutAt:
|
|
timeoutEvent?.createdAt ||
|
|
(workOrder.deadline_at ? String(workOrder.deadline_at) : null) ||
|
|
workOrder.updated_at,
|
|
warningMessage:
|
|
(workOrder.assigned_worker_id != null &&
|
|
Number(workOrder.assigned_worker_id) !== workerId) ||
|
|
['in_progress', 'submitted', 'pending_acceptance', 'accepted'].includes(
|
|
workOrder.status,
|
|
)
|
|
? '该订单已超时返回大厅并被其他打手接手，请勿继续在游戏内赠送！若私自继续赠送，造成的财产损失平台概不负责。'
|
|
: '该订单已超时退回抢单大厅，您已失去接单资格，请勿在游戏内赠送！若私自继续赠送，造成的财产损失平台概不负责。',
|
|
isTakenByOther:
|
|
(workOrder.assigned_worker_id != null &&
|
|
Number(workOrder.assigned_worker_id) !== workerId) ||
|
|
['in_progress', 'submitted', 'pending_acceptance', 'accepted'].includes(
|
|
workOrder.status,
|
|
),
|
|
}
|
|
: undefined
|
|
|
|
const hasPendingHandling =
|
|
pendingCancelOrderIds.has(workOrderId) || pendingFeedbackOrderIds.has(workOrderId)
|
|
const materialSeenAt = view?.material_seen_at || ''
|
|
const vipEvidencePending = isWorkerOrderVipEvidencePending(workOrder)
|
|
|
|
publishRealtimeEvent({
|
|
type: 'work_order.changed',
|
|
entityId: workOrderId,
|
|
scopes: ['worker_my_orders'],
|
|
operation: workerOrderVisible ? 'upsert' : 'remove',
|
|
data: {
|
|
...mapWorkOrderForWorker(
|
|
workOrder,
|
|
resolveWorkerPermissions(worker),
|
|
representativeShare,
|
|
note,
|
|
viewedAt,
|
|
{
|
|
isTimedOutForMe,
|
|
...(timeoutNotice ? { timeoutNotice } : {}),
|
|
// 与列表一致：只带未取消份额，按 id DESC（参与先后倒序）
|
|
myShares: [...activeWorkerShares].sort(
|
|
(left, right) => Number(right.id) - Number(left.id),
|
|
),
|
|
},
|
|
),
|
|
/**
|
|
* 这张单在该打手「我的订单」里属于哪些状态页签。
|
|
* 多份额工单可同时属于代练中/待验收，单个 status 字段表达不了，前端据此判归属。
|
|
*/
|
|
workerStatuses: resolveWorkerOrderStatusTabs({
|
|
workOrderStatus: workOrder.status,
|
|
isAssignedWorker,
|
|
shares: workerShares,
|
|
hasPendingHandling,
|
|
hasTimeoutForMe: Boolean(timeoutEvent),
|
|
vipEvidencePending,
|
|
}),
|
|
problemMarked:
|
|
!isTimedOutForMe &&
|
|
(workOrder.status === WORK_ORDER_STATUS.PROBLEM ||
|
|
activeWorkerShares.some((share) => Boolean(share.problem_marked_at))),
|
|
pendingCancelRequest: !isTimedOutForMe && pendingCancelOrderIds.has(workOrderId),
|
|
pendingFeedback: !isTimedOutForMe && pendingFeedbackOrderIds.has(workOrderId),
|
|
materialUpdated:
|
|
!isTimedOutForMe &&
|
|
Boolean(materialEventAt) &&
|
|
(!materialSeenAt ||
|
|
new Date(materialEventAt).getTime() > new Date(materialSeenAt).getTime()),
|
|
},
|
|
workerIds: [workerId],
|
|
})
|
|
}
|
|
}
|
|
|
|
async function publishWorkerFinanceSnapshots(input: {
|
|
requestId: number
|
|
workerId: number
|
|
walletChanged?: boolean
|
|
}) {
|
|
const [request, worker] = await Promise.all([
|
|
getWorkerFinanceRequestById(input.requestId),
|
|
getWorkerUserById(input.workerId),
|
|
])
|
|
const data = request ? mapFinanceRequest(request) : { requestId: Number(input.requestId) }
|
|
const operation = request ? 'upsert' : 'remove'
|
|
publishRealtimeEvent({
|
|
type: 'worker_finance_request.changed',
|
|
entityId: input.requestId,
|
|
scopes: ['admin_worker_finance'],
|
|
operation,
|
|
data,
|
|
admin: true,
|
|
})
|
|
publishRealtimeEvent({
|
|
type: 'worker_finance_request.changed',
|
|
entityId: input.requestId,
|
|
scopes: ['worker_finance_requests'],
|
|
operation,
|
|
data,
|
|
workerIds: [input.workerId],
|
|
})
|
|
if (input.walletChanged && worker) {
|
|
publishRealtimeEvent({
|
|
type: 'worker_wallet.changed',
|
|
entityId: input.workerId,
|
|
scopes: ['worker_wallet'],
|
|
operation: 'upsert',
|
|
data: { wallet: mapWorkerUser(worker).wallet },
|
|
workerIds: [input.workerId],
|
|
})
|
|
}
|
|
}
|
|
|
|
function openRealtimeStream(response: Response, workerId: number | null, lastEventId: number) {
|
|
const subscriber: RealtimeSubscriber = {
|
|
id: nextSubscriberId++,
|
|
response,
|
|
workerId,
|
|
}
|
|
const subscribers = workerId === null ? adminSubscribers : workerSubscribers
|
|
|
|
response.status(200)
|
|
response.setHeader('Content-Type', 'text/event-stream; charset=utf-8')
|
|
response.setHeader('Cache-Control', 'no-cache, no-transform')
|
|
response.setHeader('Connection', 'keep-alive')
|
|
response.setHeader('X-Accel-Buffering', 'no')
|
|
response.flushHeaders()
|
|
response.write('retry: 3000\n\n')
|
|
subscribers.set(subscriber.id, subscriber)
|
|
writeEvent(subscriber, 'ready', {
|
|
connectedAt: new Date().toISOString(),
|
|
lastEventId: Math.max(0, Number(lastEventId) || 0),
|
|
})
|
|
void replayEvents(subscriber, Math.max(0, Number(lastEventId) || 0))
|
|
ensureHeartbeat()
|
|
|
|
response.on('close', () => {
|
|
subscribers.delete(subscriber.id)
|
|
stopHeartbeatWhenIdle()
|
|
})
|
|
}
|
|
|
|
async function replayEvents(subscriber: RealtimeSubscriber, lastEventId: number) {
|
|
const events = await listRealtimeEventsAfter(lastEventId)
|
|
for (const row of events) {
|
|
const event = mapStoredEvent(row)
|
|
if (event.eventId <= lastEventId || !isEventVisibleToSubscriber(event, subscriber)) continue
|
|
writeEvent(subscriber, 'realtime', event)
|
|
}
|
|
}
|
|
|
|
function mapStoredEvent(row: RealtimeEventRow): StoredRealtimeEvent {
|
|
const scopes = normalizeStringArray(row.scopes)
|
|
const workerIds = normalizeStringArray(row.worker_ids)
|
|
.map((item) => Number(item))
|
|
.filter((item) => Number.isInteger(item) && item > 0)
|
|
const operation =
|
|
row.operation === 'upsert' || row.operation === 'remove' ? row.operation : ('refresh' as const)
|
|
return {
|
|
eventId: Number(row.id),
|
|
version: Number(row.id),
|
|
type: row.event_type as RealtimeEventType,
|
|
entityId: Number(row.entity_id || 0),
|
|
scopes,
|
|
operation,
|
|
...(row.data === null || row.data === undefined ? {} : { data: row.data }),
|
|
occurredAt: row.occurred_at,
|
|
adminAudience: row.admin_audience === true,
|
|
workerIds,
|
|
broadcastWorkers: row.broadcast_workers === true,
|
|
}
|
|
}
|
|
|
|
function normalizeStringArray(value: unknown): string[] {
|
|
const values =
|
|
typeof value === 'string'
|
|
? (() => {
|
|
try {
|
|
return JSON.parse(value)
|
|
} catch {
|
|
return []
|
|
}
|
|
})()
|
|
: value
|
|
return Array.isArray(values) ? values.map((item) => String(item || '')).filter(Boolean) : []
|
|
}
|
|
|
|
function isEventVisibleToSubscriber(event: StoredRealtimeEvent, subscriber: RealtimeSubscriber) {
|
|
if (subscriber.workerId === null) return event.adminAudience
|
|
return event.broadcastWorkers || event.workerIds.includes(Number(subscriber.workerId))
|
|
}
|
|
|
|
function broadcast(subscribers: Iterable<RealtimeSubscriber>, event: RealtimeEvent) {
|
|
// 同一份载荷要发给 N 个订阅者，序列化必须只做一次。
|
|
// 生产实测单个后台客户端就能收到约 67 条/秒、每条约 2.7KB 的事件，
|
|
// 原实现在订阅者循环里各写一次 JSON.stringify + 模板串，客户端越多越吃亏。
|
|
const frame = buildEventFrame('realtime', event)
|
|
for (const subscriber of subscribers) {
|
|
writeEventFrame(subscriber, frame)
|
|
}
|
|
}
|
|
|
|
/** 预先构造一整帧 SSE 报文（含 `id:` 行），供广播复用。 */
|
|
function buildEventFrame(name: string, payload: unknown): string {
|
|
const eventId =
|
|
name === 'realtime' && payload && typeof payload === 'object' && 'eventId' in payload
|
|
? Number((payload as { eventId?: unknown }).eventId || 0)
|
|
: 0
|
|
const idLine = eventId > 0 ? `id: ${eventId}\n` : ''
|
|
return `${idLine}event: ${name}\ndata: ${JSON.stringify(payload)}\n\n`
|
|
}
|
|
|
|
function writeEvent(subscriber: RealtimeSubscriber, name: string, payload: unknown) {
|
|
writeEventFrame(subscriber, buildEventFrame(name, payload))
|
|
}
|
|
|
|
function writeEventFrame(subscriber: RealtimeSubscriber, frame: string) {
|
|
if (subscriber.response.writableEnded || subscriber.response.destroyed) {
|
|
removeSubscriber(subscriber)
|
|
return
|
|
}
|
|
|
|
try {
|
|
subscriber.response.write(frame)
|
|
} catch {
|
|
removeSubscriber(subscriber)
|
|
}
|
|
}
|
|
|
|
function removeSubscriber(subscriber: RealtimeSubscriber) {
|
|
const subscribers = subscriber.workerId === null ? adminSubscribers : workerSubscribers
|
|
subscribers.delete(subscriber.id)
|
|
stopHeartbeatWhenIdle()
|
|
}
|
|
|
|
function ensureHeartbeat() {
|
|
if (heartbeatTimer) return
|
|
heartbeatTimer = setInterval(() => {
|
|
for (const subscriber of [...adminSubscribers.values(), ...workerSubscribers.values()]) {
|
|
if (subscriber.response.writableEnded || subscriber.response.destroyed) {
|
|
removeSubscriber(subscriber)
|
|
continue
|
|
}
|
|
try {
|
|
subscriber.response.write(': ping\n\n')
|
|
} catch {
|
|
removeSubscriber(subscriber)
|
|
}
|
|
}
|
|
}, 25_000)
|
|
heartbeatTimer.unref()
|
|
}
|
|
|
|
function stopHeartbeatWhenIdle() {
|
|
if (adminSubscribers.size > 0 || workerSubscribers.size > 0 || !heartbeatTimer) return
|
|
clearInterval(heartbeatTimer)
|
|
heartbeatTimer = null
|
|
}
|