fix(worker-platform): 补齐拼单份额超时自动处置机制与退回大厅流程

- 根因排查:此前超时扫描仅针对整单 in_progress,而拼单母单为 open 且 deadline_at 为空,导致打手超时的拼单份额无法被定时扫描命中与释放;
- 定时扫描:在 settleOverdueWorkOrders 中引入 listOverdueWorkOrderShares,自动扫描超时未提交资料的拼单份额(joined);
- 超时处置:新增 settleOverdueWorkOrderShare 事务处理,按母单 timeout_policy 释放或扣除押金,份额标记 cancelled,母单保持或恢复 open 状态重回抢单大厅,记录 timeout 事件并推送实时变更;
- 提交防呆:在打手提交拼单验收时检测个人份额是否已超时,超时则立即执行处置并拦截提交;
- 倒计时展示:在 mapWorkOrderForWorker 中根据 myShare.created_at + timeout_minutes 为拼单打手计算并展示个人履约截止时间;
- 质量验证:补齐拼单截止时间计算与处置链路测试,全量 676 个测试均通过,通过 npm run check 校验。
This commit is contained in:
yml2213
2026-10-11 14:53:56 +08:00
parent 06d888fb6f
commit b82f9c4063
20 changed files with 418 additions and 44 deletions
@@ -150,7 +150,7 @@ async function api<T = unknown>(
const response = await fetch(`${apiBase}${pathName}`, {
method,
headers: {
'authorization': `Bearer ${token}`,
authorization: `Bearer ${token}`,
...(body ? { 'content-type': 'application/json' } : {}),
},
body: body ? JSON.stringify(body) : undefined,
@@ -174,7 +174,6 @@ function defaultAdminUser(): { username?: string; password?: string } {
return {}
}
/**
* 打平台 messagetest:平台会**按当前订阅**把消息推回 message_url。
* 签名与消息推送同一套(我们扮演平台侧签名)。
+1 -7
View File
@@ -448,13 +448,7 @@ export async function updateOrderClaimStatus(
WHERE id = $4
RETURNING *
`,
[
input.claimStatus,
String(input.reason || '').trim(),
updatedAt,
Number(orderId),
],
[input.claimStatus, String(input.reason || '').trim(), updatedAt, Number(orderId)],
)
return result.rows[0] || null
}
@@ -326,7 +326,6 @@ export async function listWorkOrdersByOrderId(orderId: number | string): Promise
return result.rows
}
export async function listWorkOrders({
page = 1,
pageSize = 20,
@@ -731,6 +730,72 @@ export async function listOverdueWorkOrders({
return result.rows
}
export interface OverdueWorkOrderShareItem {
shareId: number
workOrderId: number
workerId: number
quantity: number
shareDeposit: number
timeoutMinutes: number
timeoutPolicy: string
shareCreatedAt: string
}
/** 查询已超过接单时限、尚未提交材料的拼单份额(status=joined)。 */
export async function listOverdueWorkOrderShares({
limit = 50,
workerId = 0,
}: { limit?: number; workerId?: number } = {}): Promise<OverdueWorkOrderShareItem[]> {
const params: unknown[] = []
let workerClause = ''
if (workerId > 0) {
params.push(workerId)
workerClause = `AND wos.worker_id = $${params.length}`
}
params.push(limit)
const result = await query<{
share_id: string
work_order_id: string
worker_id: string
quantity: number
share_deposit: number
timeout_minutes: number
timeout_policy: string
share_created_at: string
}>(
`SELECT
wos.id AS share_id,
wos.work_order_id,
wos.worker_id,
wos.quantity,
wos.share_deposit,
wo.timeout_minutes,
wo.timeout_policy,
wos.created_at AS share_created_at
FROM work_order_shares wos
INNER JOIN work_orders wo ON wo.id = wos.work_order_id
WHERE wos.status = 'joined'
AND wo.sharing_enabled = TRUE
AND wo.timeout_minutes > 0
AND wos.created_at + (wo.timeout_minutes || ' minutes')::interval < NOW()
AND wo.status = 'open'
${workerClause}
ORDER BY wos.created_at ASC
LIMIT $${params.length}`,
params,
)
return result.rows.map((row) => ({
shareId: Number(row.share_id),
workOrderId: Number(row.work_order_id),
workerId: Number(row.worker_id),
quantity: Number(row.quantity),
shareDeposit: Number(row.share_deposit),
timeoutMinutes: Number(row.timeout_minutes),
timeoutPolicy: String(row.timeout_policy || 'reopen'),
shareCreatedAt: String(row.share_created_at),
}))
}
export async function countWorkerActiveOrders(workerId: number | string): Promise<number> {
const result = await query<{ total: number }>(
`
@@ -29,4 +29,5 @@ export {
acceptWorkOrderShareAndSettle,
cancelWorkOrderShare,
cancelWorkOrderSharing,
settleOverdueWorkOrderShare,
} from './work-order-settlement-repo.js'
@@ -856,6 +856,196 @@ export async function cancelWorkOrderShare(input: {
})
}
/**
* 拼单份额超时自动处置:
* 超过母单 timeout_minutes 时限且未提交材料的 joined 份额,
* 按照母单 timeout_policy 处置(默认 reopen 释放押金退回大厅,或 cancel_deduct 扣除押金)。
*/
export async function settleOverdueWorkOrderShare(input: {
workOrderId: number
shareId: number
policy: 'reopen' | 'cancel_release' | 'cancel_deduct'
now: string
}): Promise<{
order: WorkOrderRow | null
share: WorkOrderShareRow | null
releasedDepositAmount: number
failureReason: 'work_order_not_cancellable' | 'share_not_found' | 'share_not_overdue' | null
}> {
return withTransaction(async (client) => {
const orderResult = await client.query<WorkOrderRow>(
`SELECT * FROM work_orders WHERE id = $1 FOR UPDATE`,
[input.workOrderId],
)
const workOrder = orderResult.rows[0] || null
if (!workOrder || workOrder.sharing_enabled !== true || workOrder.status !== 'open') {
return {
order: workOrder,
share: null,
releasedDepositAmount: 0,
failureReason: 'work_order_not_cancellable',
}
}
const shareResult = await client.query<WorkOrderShareRow>(
`SELECT * FROM work_order_shares WHERE id = $1 AND work_order_id = $2 FOR UPDATE`,
[input.shareId, input.workOrderId],
)
const share = shareResult.rows[0] || null
if (!share || share.status !== 'joined') {
return {
order: workOrder,
share,
releasedDepositAmount: 0,
failureReason: 'share_not_found',
}
}
const timeoutMinutes = Number(workOrder.timeout_minutes || 0)
if (timeoutMinutes <= 0) {
return {
order: workOrder,
share,
releasedDepositAmount: 0,
failureReason: 'share_not_overdue',
}
}
const deadlineMs = new Date(share.created_at).getTime() + timeoutMinutes * 60 * 1000
if (deadlineMs > new Date(input.now).getTime()) {
return {
order: workOrder,
share,
releasedDepositAmount: 0,
failureReason: 'share_not_overdue',
}
}
const workerId = Number(share.worker_id || 0)
let releasedDepositAmount = 0
let deductedDepositAmount = 0
if (workerId > 0) {
await ensureWorkerWalletWithClient(client, workerId, input.now)
const wallet = await getWorkerWalletWithClient(client, workerId)
const frozenDeposit = Math.min(
Number(share.share_deposit || 0),
Number(wallet?.frozen_deposit_amount || 0),
)
if (frozenDeposit > 0) {
if (input.policy === 'cancel_deduct') {
deductedDepositAmount = frozenDeposit
const nextFrozen = Math.max(0, Number(wallet?.frozen_deposit_amount || 0) - frozenDeposit)
await client.query(
`UPDATE worker_wallets
SET frozen_deposit_amount = $1, updated_at = $2
WHERE worker_id = $3`,
[nextFrozen, input.now, workerId],
)
await client.query(
`INSERT INTO worker_wallet_ledgers (
worker_id, ledger_type, amount, balance_after, frozen_after,
related_work_order_id, note, payload_json, created_at
) VALUES ($1, 'deposit_deduction', $2, $3, $4, $5, '拼单超时处置扣除押金', $6::jsonb, $7)`,
[
workerId,
-frozenDeposit,
Number(wallet?.available_amount || 0),
nextFrozen,
input.workOrderId,
JSON.stringify({
workOrderId: input.workOrderId,
shareId: input.shareId,
action: 'timeout_share_deduct',
}),
input.now,
],
)
} else {
// reopen or cancel_release
releasedDepositAmount = frozenDeposit
const nextAvailable = Number(wallet?.available_amount || 0) + frozenDeposit
const nextFrozen = Math.max(0, Number(wallet?.frozen_deposit_amount || 0) - frozenDeposit)
await client.query(
`UPDATE worker_wallets
SET available_amount = $1, frozen_deposit_amount = $2, updated_at = $3
WHERE worker_id = $4`,
[nextAvailable, nextFrozen, input.now, workerId],
)
await client.query(
`INSERT INTO worker_wallet_ledgers (
worker_id, ledger_type, amount, balance_after, frozen_after,
related_work_order_id, note, payload_json, created_at
) VALUES ($1, 'deposit_release', $2, $3, $4, $5, '拼单超时退回大厅释放押金', $6::jsonb, $7)`,
[
workerId,
frozenDeposit,
nextAvailable,
nextFrozen,
input.workOrderId,
JSON.stringify({
workOrderId: input.workOrderId,
shareId: input.shareId,
action: 'timeout_share_reopen',
}),
input.now,
],
)
}
}
}
await client.query(
`UPDATE work_order_shares
SET status = 'cancelled',
acceptance_json = '{}'::jsonb,
draft_acceptance_json = '{}'::jsonb,
submitted_at = NULL,
accepted_at = NULL,
updated_at = $1
WHERE id = $2`,
[input.now, input.shareId],
)
await client.query(
`UPDATE work_orders
SET status = 'open',
published_at = COALESCE(published_at, $1),
hall_queued_at = $1,
updated_at = $1
WHERE id = $2`,
[input.now, input.workOrderId],
)
await createWorkOrderEventWithClient(client, {
workOrderId: input.workOrderId,
actorType: 'system',
actorId: '',
eventType: `timeout_${input.policy}`,
fromStatus: 'open',
toStatus: 'open',
payloadJson: JSON.stringify({
policy: input.policy,
workerId,
shareId: input.shareId,
quantity: Number(share.quantity || 0),
depositAmount: releasedDepositAmount || deductedDepositAmount,
isSharing: true,
}),
now: input.now,
})
const updatedShareResult = await client.query<WorkOrderShareRow>(
`${WORK_ORDER_SHARE_SELECT} WHERE wos.id = $1 LIMIT 1`,
[input.shareId],
)
return {
order: await getWorkOrderByIdWithClient(client, input.workOrderId),
share: updatedShareResult.rows[0] || null,
releasedDepositAmount,
failureReason: null,
}
})
}
/**
* 母单级拼单撤单。只撤回尚未验收的 joined/submitted 份额;已验收份额保留,
* 取消整单时则要求不存在已验收份额,避免绕过售后和已结算资金处理。
-1
View File
@@ -86,4 +86,3 @@ router.post(
)
export default router
@@ -82,4 +82,3 @@ router.post(
)
export default router
@@ -128,7 +128,9 @@ export function resolveOrderFulfillmentType({
fulfillmentType: 'auto_claim' | 'worker_dispatch' | 'voucher_only' | 'manual'
canClaimH5: boolean
} {
const normProvider = String(provider || '').trim().toLowerCase()
const normProvider = String(provider || '')
.trim()
.toLowerCase()
const hasAutoClaimTask =
tasks.some(
(t) =>
@@ -139,7 +139,9 @@ export async function getAdminOrders(
}
}
function groupRowsByOrderId<T extends { order_id?: number | string | null }>(rows: T[]): Map<number, T[]> {
function groupRowsByOrderId<T extends { order_id?: number | string | null }>(
rows: T[],
): Map<number, T[]> {
const grouped = new Map<number, T[]>()
for (const row of rows) {
// pg BIGINT 常以 string 返回;统一 Number 键,避免列表挂不上 order_items →「商品 0」。
@@ -292,7 +292,9 @@ function resolveTaskDisplayRoleName(task: TaskRow): string {
const affiliateDash = context.affiliateDash
if (affiliateDash && typeof affiliateDash === 'object') {
const dash = affiliateDash as Record<string, unknown>
const name = String(dash.roleName || dash.boundRoleName || dash.game_account_role_name || '').trim()
const name = String(
dash.roleName || dash.boundRoleName || dash.game_account_role_name || '',
).trim()
if (name) return name
}
const kuaishouCloud = context.kuaishouCloudFulfillment
@@ -314,11 +316,11 @@ function resolveTaskDisplayRoleId(task: TaskRow): string {
const dash = affiliateDash as Record<string, unknown>
const id = String(
dash.boundAccount ||
dash.gameAccount ||
dash.expectedGameAccount ||
dash.game_uid ||
dash.game_account ||
'',
dash.gameAccount ||
dash.expectedGameAccount ||
dash.game_uid ||
dash.game_account ||
'',
).trim()
if (id) return id
}
@@ -184,4 +184,3 @@ export function isMaskedSecretValue(value: unknown, currentSecret = ''): boolean
}
return false
}
@@ -1,12 +1,8 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import {
getOrderClaimContext,
} from './kuaishou-cloud-claim-context.js'
import {
updateAdminOrderClaimStatus,
} from '../admin/write/order-actions.js'
import { getOrderClaimContext } from './kuaishou-cloud-claim-context.js'
import { updateAdminOrderClaimStatus } from '../admin/write/order-actions.js'
test('getOrderClaimContext: 缺少订单号时抛出 400 错误', async () => {
await assert.rejects(
@@ -1,7 +1,11 @@
import { findClaimTokenByToken, updateClaimToken } from '../../repositories/claim-token-repo.js'
import { getOrderItemById, listOrderItemsByOrderId } from '../../repositories/order-item-repo.js'
import { findLatestOrderByAnyPlatformOrderId, getOrderById } from '../../repositories/order-repo.js'
import { findTaskByClaimTokenId, listTasksByOrderId, updateTask } from '../../repositories/task-repo.js'
import {
findTaskByClaimTokenId,
listTasksByOrderId,
updateTask,
} from '../../repositories/task-repo.js'
import { createHttpError } from '../../utils/http.js'
import { formatFenToAmount, normalizeFen } from '../../utils/money.js'
import { parseTaskContext as parseTaskContextValue } from '../../utils/task-json.js'
@@ -28,7 +32,6 @@ export type ClaimContext = {
orderItem: OrderItemRow
}
export async function getClaimContext(token: unknown) {
const normalized = String(token || '').trim()
@@ -194,9 +197,7 @@ export async function getOrderClaimContext(orderNo: unknown): Promise<ClaimConte
})
}
const activeTask =
tasks.find((t) => !isTaskFinalStatus(t.task_status)) ||
tasks[tasks.length - 1]
const activeTask = tasks.find((t) => !isTaskFinalStatus(t.task_status)) || tasks[tasks.length - 1]
if (!activeTask) {
throw createHttpError('该订单暂无可用履约任务', {
@@ -205,9 +206,7 @@ export async function getOrderClaimContext(orderNo: unknown): Promise<ClaimConte
})
}
const orderItem =
items.find((it) => it.id === activeTask.order_item_id) ||
items[0]
const orderItem = items.find((it) => it.id === activeTask.order_item_id) || items[0]
if (!orderItem) {
throw createHttpError('订单商品明细缺失', {
@@ -227,7 +226,9 @@ export async function getOrderClaimContext(orderNo: unknown): Promise<ClaimConte
const claimToken: ClaimTokenRow = {
id: Number(activeTask.primary_claim_token_id || 0),
token: String(activeTask.primary_claim_token || activeTask.claim_token || `h5_${order.platform_order_id}`).trim(),
token: String(
activeTask.primary_claim_token || activeTask.claim_token || `h5_${order.platform_order_id}`,
).trim(),
status: 'active',
expired_at: activeTask.claim_expires_at || null,
created_at: activeTask.created_at,
@@ -177,4 +177,3 @@ export async function configureDouyinMashidaMessage(
function isPlainObject(value: unknown): value is JsonObject {
return Object.prototype.toString.call(value) === '[object Object]'
}
@@ -246,7 +246,9 @@ async function notifyBuyerClaimLink(input: {
if (input.config.claimMessageEnabled === false) {
return
}
const claimLinkBase = String(input.config.claimLinkBase || '').trim().replace(/\/+$/, '')
const claimLinkBase = String(input.config.claimLinkBase || '')
.trim()
.replace(/\/+$/, '')
if (!claimLinkBase) {
return
}
@@ -140,7 +140,9 @@ export async function resolveDouyinMashidaCertLink(input: {
config: DouyinMashidaSourceConfig
}): Promise<string> {
const fallback = String(input.config.certLink || '').trim()
const claimLinkBase = String(input.config.claimLinkBase || '').trim().replace(/\/+$/, '')
const claimLinkBase = String(input.config.claimLinkBase || '')
.trim()
.replace(/\/+$/, '')
if (!claimLinkBase) {
return fallback
}
@@ -457,6 +457,12 @@ export function mapWorkOrderForWorker(
assignedAt: myShare?.created_at || visibleOrder.assignedAt,
submittedAt: myShare?.submitted_at || visibleOrder.submittedAt,
acceptedAt: myShare?.accepted_at || visibleOrder.acceptedAt,
deadlineAt:
myShare && Number(workOrder.timeout_minutes || 0) > 0
? new Date(
new Date(myShare.created_at).getTime() + Number(workOrder.timeout_minutes) * 60 * 1000,
).toISOString()
: visibleOrder.deadlineAt,
workerNote,
myShare: myShare ? mapWorkOrderShare(myShare) : null,
...(viewedAt === undefined ? {} : { isUnread: !viewedAt }),
@@ -11,6 +11,7 @@ import {
createWorkOrderEvent,
getWorkOrderShare,
settleOverdueWorkOrder,
settleOverdueWorkOrderShare,
submitWorkOrderShareAcceptance,
updateWorkOrder,
updateWorkOrderAcceptanceRecord,
@@ -410,6 +411,29 @@ async function submitWorkerSharingAcceptance(
})
}
const now = nowIso()
const timeoutMinutes = Number(workOrder.timeout_minutes || 0)
if (
sharingShare.status === 'joined' &&
timeoutMinutes > 0 &&
new Date(sharingShare.created_at).getTime() + timeoutMinutes * 60 * 1000 <= Date.now()
) {
await settleOverdueWorkOrderShare({
workOrderId: workOrder.id,
shareId: sharingShare.id,
policy: normalizeWorkOrderTimeoutPolicy(workOrder.timeout_policy),
now,
})
publishWorkOrderRealtimeChange({
workOrderId: Number(workOrder.id),
workerIds: [Number(session.workerId)],
hallChanged: true,
})
publishWorkerWalletRealtimeChange(Number(session.workerId))
throw createHttpError('拼单任务已超时,系统已按超时策略处置', {
statusCode: 409,
errorCode: 'work_order_timeout_expired',
})
}
const worker = await getRequiredWorker(session.workerId)
const canAutoAcceptWithoutEvidence = resolveWorkerPermissions(worker).autoAcceptWithoutEvidence
const { files, imageUrls, note } = resolveAcceptanceImages(
@@ -1,7 +1,9 @@
import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js'
import {
listOverdueWorkOrders,
listOverdueWorkOrderShares,
settleOverdueWorkOrder,
settleOverdueWorkOrderShare,
} from '../../repositories/worker-platform/index.js'
import { nowIso } from '../../utils/time.js'
import {
@@ -11,10 +13,10 @@ import {
import { normalizeWorkOrderTimeoutPolicy } from './work-order-timeout-policy.js'
export async function settleOverdueWorkOrders(options: { limit?: number; workerId?: number } = {}) {
const overdue = await listOverdueWorkOrders({
limit: Math.max(1, Number(options.limit || 50)),
workerId: Number(options.workerId || 0),
})
const limit = Math.max(1, Number(options.limit || 50))
const workerId = Number(options.workerId || 0)
const overdue = await listOverdueWorkOrders({ limit, workerId })
let processedCount = 0
let skippedCount = 0
for (const workOrder of overdue) {
@@ -39,8 +41,33 @@ export async function settleOverdueWorkOrders(options: { limit?: number; workerI
skippedCount += 1
}
}
// 处置超时未提交材料的拼单份额(status=joined)
const overdueShares = await listOverdueWorkOrderShares({ limit, workerId })
for (const shareItem of overdueShares) {
const result = await settleOverdueWorkOrderShare({
workOrderId: shareItem.workOrderId,
shareId: shareItem.shareId,
policy: normalizeWorkOrderTimeoutPolicy(shareItem.timeoutPolicy),
now: nowIso(),
})
if (result.share) {
processedCount += 1
publishWorkOrderRealtimeChange({
workOrderId: Number(shareItem.workOrderId),
workerIds: Number(shareItem.workerId) ? [Number(shareItem.workerId)] : [],
hallChanged: true,
})
if (Number(shareItem.workerId)) {
publishWorkerWalletRealtimeChange(Number(shareItem.workerId))
}
} else {
skippedCount += 1
}
}
return {
checkedCount: overdue.length,
checkedCount: overdue.length + overdueShares.length,
processedCount,
skippedCount,
}
@@ -3059,3 +3059,68 @@ test('已冻结打手调级或变更资料不应因未传 freezeReason 报错',
'必须区分是否为新冻结,防止已冻结账号调级时抛出 freeze_reason 必填错误',
)
})
test('mapWorkOrderForWorker 拼单工单正确计算个人份额截止时间', () => {
const workOrder = buildWorkOrderRow({
product_name: '指挥官秘钥15个',
status: 'open',
sharing_enabled: true,
sharing_total_quantity: 15,
sharing_unit_reward: 700,
timeout_minutes: 120,
deadline_at: null,
})
const share = buildWorkOrderShareRow({
created_at: '2026-08-18T09:09:08.000Z',
})
const mapped = mapWorkOrderForWorker(
workOrder,
resolveWorkerPermissions(buildWorkerUserRow()),
share,
)
assert.equal(mapped.deadlineAt, '2026-08-18T11:09:08.000Z')
})
test('拼单超时处置链路完整性验证', async () => {
const { readFile } = await import('node:fs/promises')
const timeoutSrc = await readFile(
new URL('./worker-order-timeout-service.ts', import.meta.url),
'utf8',
)
assert.ok(
timeoutSrc.includes('listOverdueWorkOrderShares') &&
timeoutSrc.includes('settleOverdueWorkOrderShare'),
'settleOverdueWorkOrders 必须同时扫描并处置超时拼单份额',
)
const acceptanceSrc = await readFile(
new URL('./worker-acceptance-service.ts', import.meta.url),
'utf8',
)
assert.ok(
acceptanceSrc.includes('settleOverdueWorkOrderShare') &&
acceptanceSrc.includes('work_order_timeout_expired'),
'submitWorkerSharingAcceptance 必须在提交验收时拦截超时份额并处置',
)
const queryRepoSrc = await readFile(
new URL('../../repositories/worker-platform/work-order-query-repo.ts', import.meta.url),
'utf8',
)
assert.ok(
queryRepoSrc.includes('listOverdueWorkOrderShares') &&
queryRepoSrc.includes("wos.status = 'joined'"),
'listOverdueWorkOrderShares 必须查询超时且未提交材料的 joined 份额',
)
const settlementRepoSrc = await readFile(
new URL('../../repositories/worker-platform/work-order-settlement-repo.ts', import.meta.url),
'utf8',
)
assert.ok(
settlementRepoSrc.includes('settleOverdueWorkOrderShare') &&
settlementRepoSrc.includes("SET status = 'cancelled'"),
'settleOverdueWorkOrderShare 必须将超时份额置为 cancelled 并释放/扣除押金',
)
})