perf: 筛选统计去掉 JSON 兜底,并补上 orders 覆盖索引
去掉 work_orders 堆访问后,JOIN 会把 orders 整表扫进 Hash Right Join。 领取页在快照新鲜时不再打上游;快手出网统一 10 秒超时。倒计时改为自适应间隔。 Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
-- 087_work_order_statistics_cover_index.sql —— 筛选统计的覆盖索引(085 的超集 + orders 窄列)。
|
||||
--
|
||||
-- 统计在 SUM 付款金额,085 只覆盖了运营态 CASE 的三列,金额表达式此前还引用
|
||||
-- material_json,规划器因此无法 index-only scan(缓存未命中实测 248–545MB)。
|
||||
-- 金额已改为只读 source_order_amount_fen / orders.total_amount。
|
||||
--
|
||||
-- 生产同一会话对比(唯一变量是表达式,不勾 VIP 待补图):
|
||||
-- 旧(含 JSON) work_orders Parallel Seq Scan,合计 550MB / 615ms
|
||||
-- 只加 work_orders 索引 work_orders Index Only Scan 56ms,
|
||||
-- 但 LEFT JOIN orders 变成 Hash Right Join,
|
||||
-- 26.5 万行 orders 整表 Seq Scan 235MB,合计 487ms
|
||||
-- 再加下面这条 orders 索引 两侧都是 Index Only Scan,合计 48MB / 317ms
|
||||
-- 所以 orders 索引必须跟这批一起建。生产上已并发建好,IF NOT EXISTS 在线上是 no-op;
|
||||
-- 不写进迁移的话,新环境会退回 orders 全表扫。
|
||||
--
|
||||
-- idx_orders_statistics_cover 与 orders_pkey 同键,只多 3 个 INCLUDE 列(约 10MB)。
|
||||
-- 彻底去重要重建 PRIMARY KEY (id) INCLUDE (...),会锁 443MB 的 orders,留维护窗口。
|
||||
-- 现在先接受这条重复索引。
|
||||
--
|
||||
-- 验收(真实 service 路径的 EXPLAIN (ANALYZE, BUFFERS),不勾 VIP 待补图):
|
||||
-- 1. 两侧都是 Index Only Scan。Heap Fetches 不可能为 0:VACUUM ANALYZE 之后
|
||||
-- 仍有约 1.7%(1,616 / 1,635)来自刚写入、可见性映射还没覆盖的行。
|
||||
-- 判据是低于 5%,或绝对数不超过数千。明显更高时再 VACUUM (ANALYZE)。
|
||||
-- 2. 单次耗时低于 500ms 慢查询线即可。实测约 317ms 是 CPU
|
||||
-- (约 9.7 万行两次 hash join + CASE 聚合),再往下压靠预聚合,不靠索引。
|
||||
--
|
||||
-- 故意不含 acceptance_json:勾「VIP 待补图」时统计会回表,这是有意降级。
|
||||
-- 把该列补进 INCLUDE 会让默认统计重新读宽表。
|
||||
|
||||
CREATE INDEX IF NOT EXISTS idx_work_orders_statistics_cover
|
||||
ON work_orders (id)
|
||||
INCLUDE (
|
||||
status,
|
||||
sharing_enabled,
|
||||
sharing_total_quantity,
|
||||
source_order_amount_fen,
|
||||
reward_amount,
|
||||
order_id
|
||||
);
|
||||
|
||||
-- 见文件头:挡住 orders 被当成 Hash Right Join 的 build 侧整表扫。
|
||||
CREATE INDEX IF NOT EXISTS idx_orders_statistics_cover
|
||||
ON orders (id)
|
||||
INCLUDE (pay_status, order_status, total_amount);
|
||||
|
||||
DROP INDEX IF EXISTS idx_work_orders_operational_status_cover;
|
||||
@@ -333,31 +333,26 @@ async function countWorkOrdersWithWhere(
|
||||
|
||||
/** 按筛选条件汇总工单,统计范围不受当前分页影响。 */
|
||||
/**
|
||||
* 统计用的「用户付款金额」表达式。
|
||||
* 统计用的「用户付款金额」。
|
||||
*
|
||||
* 优先读 083 迁移落下来的派生列 `source_order_amount_fen`(写入点:createWorkOrder /
|
||||
* updateWorkOrder,口径见 shared.ts 的 resolveSourceOrderAmountFen)。
|
||||
* **COALESCE 是短路求值**:列非空时后面两个分支根本不会被计算,所以命中列的行完全不碰
|
||||
* material_json——这正是把 2.8 秒的统计查询拉回毫秒级的关键(改造前每行要解析 1.2KB JSON,
|
||||
* 全表约 132MB,而且这个表达式在总付款与退款两处各求值一次)。
|
||||
* 只读 083 的派生列 `source_order_amount_fen`(写入点:createWorkOrder / updateWorkOrder,
|
||||
* 口径见 shared.ts 的 resolveSourceOrderAmountFen)。NULL 表示来源没给合法整数,
|
||||
* 回退 `orders.total_amount`,与 083 的列注释一致。
|
||||
*
|
||||
* 第二、三个分支是兜底,只为覆盖"写入点漏了"或迁移前的老行,保证口径永远正确:
|
||||
* 1) 派生列为 NULL 且 JSON 里有 15 位以内的合法整数 → 用 JSON 值;
|
||||
* 2) 否则回退 orders.total_amount(与改造前完全一致)。
|
||||
* 这里不能再写 `material_json`。执行器对 COALESCE 会短路,但规划器在规划期就记下列引用,
|
||||
* 只要 SQL 文本里出现这列,覆盖索引就做不了 index-only scan,9.7 万行宽表仍会被堆访问
|
||||
* (生产实测缓存未命中 248–545MB / 约 400ms)。漏写不在热路径上用 JSON 兜底,
|
||||
* 交给数据保留任务回填并打日志。
|
||||
*/
|
||||
const WORK_ORDER_PAYMENT_AMOUNT_SQL = `
|
||||
COALESCE(
|
||||
wo.source_order_amount_fen,
|
||||
CASE
|
||||
WHEN COALESCE(wo.material_json->'source'->>'orderAmountFen', '') ~ '^-?[0-9]{1,15}$'
|
||||
THEN (wo.material_json->'source'->>'orderAmountFen')::bigint
|
||||
END,
|
||||
COALESCE(o.total_amount, 0)
|
||||
)
|
||||
COALESCE(wo.source_order_amount_fen, o.total_amount, 0)
|
||||
`
|
||||
|
||||
export async function getWorkOrderStatistics(input: ListInput = {}): Promise<WorkOrderStatistics> {
|
||||
const { joins, whereClause, params } = buildWorkOrderWhere(input)
|
||||
// 仅勾「VIP 待补图」时才拼进 SQL。087 的覆盖索引故意不含 acceptance_json:
|
||||
// 这个筛选项会让统计回表变慢,是有意降级。不要把 acceptance_json 补进索引,
|
||||
// 否则默认统计(不勾这个筛选)会退回宽表堆访问。
|
||||
const vipEvidencePendingWhere = `
|
||||
wo.status = 'accepted'
|
||||
AND wo.acceptance_json->>'reviewMode' = 'vip_auto'
|
||||
|
||||
@@ -2,8 +2,11 @@ import test from 'node:test'
|
||||
import assert from 'node:assert/strict'
|
||||
|
||||
import {
|
||||
AFFILIATE_DASH_BIND_REFRESH_FRESH_MS,
|
||||
AFFILIATE_DASH_ORDER_SYNC_FRESH_MS,
|
||||
canReuseAffiliateDashBindSnapshot,
|
||||
isAffiliateDashBindSnapshotStale,
|
||||
isAffiliateDashSnapshotFresh,
|
||||
shouldRefreshAffiliateDashBindStateForSnapshot,
|
||||
} from './kuaishou-cloud-claim-service.js'
|
||||
|
||||
@@ -138,6 +141,84 @@ test('canReuseAffiliateDashBindSnapshot refuses mismatched bound account', () =>
|
||||
)
|
||||
})
|
||||
|
||||
test('isAffiliateDashSnapshotFresh treats missing or future timestamps as stale', () => {
|
||||
const now = Date.parse('2026-10-01T06:00:00.000Z')
|
||||
assert.equal(isAffiliateDashSnapshotFresh(null, AFFILIATE_DASH_ORDER_SYNC_FRESH_MS, now), false)
|
||||
assert.equal(isAffiliateDashSnapshotFresh('', AFFILIATE_DASH_ORDER_SYNC_FRESH_MS, now), false)
|
||||
assert.equal(
|
||||
isAffiliateDashSnapshotFresh('not-a-date', AFFILIATE_DASH_ORDER_SYNC_FRESH_MS, now),
|
||||
false,
|
||||
)
|
||||
assert.equal(
|
||||
isAffiliateDashSnapshotFresh(
|
||||
'2026-10-01T06:00:01.000Z',
|
||||
AFFILIATE_DASH_ORDER_SYNC_FRESH_MS,
|
||||
now,
|
||||
),
|
||||
false,
|
||||
)
|
||||
})
|
||||
|
||||
test('order snapshot stays fresh for 45s and bind snapshot only for 15s', () => {
|
||||
const now = Date.parse('2026-10-01T06:00:45.000Z')
|
||||
assert.equal(
|
||||
isAffiliateDashSnapshotFresh(
|
||||
'2026-10-01T06:00:00.000Z',
|
||||
AFFILIATE_DASH_ORDER_SYNC_FRESH_MS,
|
||||
now,
|
||||
),
|
||||
false,
|
||||
)
|
||||
assert.equal(
|
||||
isAffiliateDashSnapshotFresh(
|
||||
'2026-10-01T06:00:01.000Z',
|
||||
AFFILIATE_DASH_ORDER_SYNC_FRESH_MS,
|
||||
now,
|
||||
),
|
||||
true,
|
||||
)
|
||||
assert.equal(
|
||||
isAffiliateDashSnapshotFresh(
|
||||
'2026-10-01T06:00:30.000Z',
|
||||
AFFILIATE_DASH_BIND_REFRESH_FRESH_MS,
|
||||
now,
|
||||
),
|
||||
false,
|
||||
)
|
||||
assert.equal(
|
||||
isAffiliateDashSnapshotFresh(
|
||||
'2026-10-01T06:00:31.000Z',
|
||||
AFFILIATE_DASH_BIND_REFRESH_FRESH_MS,
|
||||
now,
|
||||
),
|
||||
true,
|
||||
)
|
||||
})
|
||||
|
||||
test('claim detail skips affiliate-dash sync while the snapshot is fresh', async () => {
|
||||
const { readFile } = await import('node:fs/promises')
|
||||
const src = await readFile(new URL('./kuaishou-cloud-claim-service.ts', import.meta.url), 'utf8')
|
||||
const detail = src.slice(
|
||||
src.indexOf('export async function getKuaishouCloudClaimDetail'),
|
||||
src.indexOf('export async function submitClaimUid'),
|
||||
)
|
||||
assert.ok(detail.includes('flow.mock?.enabled'))
|
||||
assert.ok(detail.includes('isAffiliateDashSnapshotFresh(flow.lastSyncedAt'))
|
||||
assert.ok(detail.includes('bindFlow.lastBindSyncedAt'))
|
||||
assert.ok(
|
||||
!detail.includes('snapshotOnly'),
|
||||
'新鲜快照必须直接不调用同步;snapshotOnly 仍会 updateTask',
|
||||
)
|
||||
const refresh = src.slice(
|
||||
src.indexOf('async function refreshAffiliateDashBindState'),
|
||||
src.indexOf('function normalizeUid'),
|
||||
)
|
||||
assert.ok(
|
||||
refresh.includes('lastBindSyncedAt: now'),
|
||||
'成功拉取绑定结果后才记时间,失败不能挡住下一次',
|
||||
)
|
||||
})
|
||||
|
||||
test('shouldRefreshAffiliateDashBindStateForSnapshot keeps polling before submit', () => {
|
||||
assert.equal(
|
||||
shouldRefreshAffiliateDashBindStateForSnapshot({
|
||||
|
||||
@@ -26,6 +26,26 @@ type ClaimDetailPayload = ReturnType<typeof buildClaimDetailPayload>
|
||||
|
||||
const RETIRED_EXECUTOR_KEYS = new Set(['kuaishou_feifei', 'kuaishou_ct_assisted'])
|
||||
|
||||
/**
|
||||
* 订单状态轮询的新鲜窗口。
|
||||
* 发货状态以 affiliate-dash notify 回调为主(notify-service 会调 syncAffiliateDashTaskStatus)。
|
||||
* 领取页详情只在回调没到时做备份:窗口内不打上游、也不写库。
|
||||
* 代价是回调丢失时,「已发货」最多晚这么久才出现在领取页上。
|
||||
*/
|
||||
export const AFFILIATE_DASH_ORDER_SYNC_FRESH_MS = 45_000
|
||||
/** 绑定结果没有回调,扫码过程仍要靠轮询,但必须长于上游超时,避免 8 秒一轮必超时。 */
|
||||
export const AFFILIATE_DASH_BIND_REFRESH_FRESH_MS = 15_000
|
||||
|
||||
export function isAffiliateDashSnapshotFresh(
|
||||
syncedAt: unknown,
|
||||
freshForMs: number,
|
||||
now = Date.now(),
|
||||
) {
|
||||
const parsed = Date.parse(String(syncedAt ?? ''))
|
||||
if (!Number.isFinite(parsed)) return false
|
||||
return now - parsed >= 0 && now - parsed < freshForMs
|
||||
}
|
||||
|
||||
function isRetiredExecutor(executorKey: unknown) {
|
||||
return RETIRED_EXECUTOR_KEYS.has(String(executorKey || '').trim())
|
||||
}
|
||||
@@ -46,9 +66,25 @@ export async function getKuaishouCloudClaimDetail(token: unknown): Promise<Claim
|
||||
}
|
||||
|
||||
if (executorKey === 'affiliate_dash') {
|
||||
task = (await syncAffiliateDashTaskStatus(task)) || task
|
||||
const flow = normalizeAffiliateDashFlow(parseTaskContextValue(task).affiliateDash)
|
||||
// 新鲜则整段不调用。走同步函数的本地分支仍会 updateTask,8 秒一轮照样写 WAL。
|
||||
// dev-mock 不打上游,但任务状态要靠这次调用从快照落下来,不能被新鲜窗口跳过。
|
||||
if (
|
||||
flow.mock?.enabled ||
|
||||
!isAffiliateDashSnapshotFresh(flow.lastSyncedAt, AFFILIATE_DASH_ORDER_SYNC_FRESH_MS)
|
||||
) {
|
||||
task = (await syncAffiliateDashTaskStatus(task)) || task
|
||||
}
|
||||
if (shouldRefreshAffiliateDashBindState(task)) {
|
||||
task = (await refreshAffiliateDashBindState(task)) || task
|
||||
const bindFlow = normalizeAffiliateDashFlow(parseTaskContextValue(task).affiliateDash)
|
||||
if (
|
||||
!isAffiliateDashSnapshotFresh(
|
||||
bindFlow.lastBindSyncedAt,
|
||||
AFFILIATE_DASH_BIND_REFRESH_FRESH_MS,
|
||||
)
|
||||
) {
|
||||
task = (await refreshAffiliateDashBindState(task)) || task
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@@ -564,6 +600,7 @@ async function refreshAffiliateDashBindState(task: TaskRow): Promise<TaskRow | n
|
||||
expectedGameAccount &&
|
||||
normalizeUid(boundAccount) !== normalizeUid(expectedGameAccount),
|
||||
)
|
||||
const now = nowIso()
|
||||
const nextFlow = {
|
||||
...flow,
|
||||
bindUuid: result.bindUuid || flow.bindUuid,
|
||||
@@ -580,8 +617,8 @@ async function refreshAffiliateDashBindState(task: TaskRow): Promise<TaskRow | n
|
||||
? String(result.gameChannel || flow.gameChannel || '').trim()
|
||||
: flow.gameChannel,
|
||||
bindMismatch: Boolean(result.mismatch) || localMismatch,
|
||||
lastBindSyncedAt: now,
|
||||
}
|
||||
const now = nowIso()
|
||||
|
||||
return (
|
||||
(await updateTask(task.id, {
|
||||
|
||||
@@ -330,6 +330,8 @@ export type AffiliateDashFlow = {
|
||||
submitStatus: string
|
||||
consumeStatus: string
|
||||
lastSyncedAt: unknown
|
||||
/** 最近一次成功拉取 bind-result 的时间。与订单状态的 lastSyncedAt 分开计。 */
|
||||
lastBindSyncedAt: unknown
|
||||
raw: unknown
|
||||
/** dev-mock 标记:enabled 时所有 affiliate-dash 平台调用短路(建单/绑定/提交/同步均不请求外部) */
|
||||
mock: { enabled: boolean; orderNo: string; createdAt: string } | null
|
||||
@@ -371,6 +373,7 @@ export function normalizeAffiliateDashFlow(value: unknown): AffiliateDashFlow {
|
||||
submitStatus: String(source.submitStatus || '').trim(),
|
||||
consumeStatus: String(source.consumeStatus || 'pending').trim(),
|
||||
lastSyncedAt: source.lastSyncedAt || null,
|
||||
lastBindSyncedAt: source.lastBindSyncedAt || null,
|
||||
raw,
|
||||
mock:
|
||||
source.mock && typeof source.mock === 'object'
|
||||
|
||||
@@ -0,0 +1,16 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import { readFile } from 'node:fs/promises'
|
||||
import test from 'node:test'
|
||||
|
||||
test('快手行业出网调用都走带超时的 fetch 封装', async () => {
|
||||
const http = await readFile(new URL('./http.ts', import.meta.url), 'utf8')
|
||||
assert.ok(http.includes('KUAISHOU_INDUSTRY_HTTP_TIMEOUT_MS = 10_000'))
|
||||
assert.ok(http.includes('AbortSignal.timeout(KUAISHOU_INDUSTRY_HTTP_TIMEOUT_MS)'))
|
||||
assert.ok(http.includes('init?.signal ??'), '调用方已经传入 signal 时不要覆盖,避免打断主动取消')
|
||||
|
||||
for (const file of ['./openapi-client.ts', './token-service.ts']) {
|
||||
const src = await readFile(new URL(file, import.meta.url), 'utf8')
|
||||
assert.ok(src.includes('fetchKuaishouIndustry('), `${file} 仍在直接 fetch`)
|
||||
assert.equal(src.includes('await fetch('), false, `${file} 还有未收口的 fetch`)
|
||||
}
|
||||
})
|
||||
@@ -0,0 +1,12 @@
|
||||
/** 快手行业出网统一入口。超时收在这里,避免每个调用点各写各的 fetch 再漏掉。 */
|
||||
export const KUAISHOU_INDUSTRY_HTTP_TIMEOUT_MS = 10_000
|
||||
|
||||
export function fetchKuaishouIndustry(
|
||||
input: Parameters<typeof fetch>[0],
|
||||
init?: Parameters<typeof fetch>[1],
|
||||
) {
|
||||
return fetch(input, {
|
||||
...init,
|
||||
signal: init?.signal ?? AbortSignal.timeout(KUAISHOU_INDUSTRY_HTTP_TIMEOUT_MS),
|
||||
})
|
||||
}
|
||||
@@ -1,5 +1,6 @@
|
||||
import { getKuaishouIndustryConfig } from './config.js'
|
||||
import { signKuaishouIndustryPayload, type SignMethod } from './crypto.js'
|
||||
import { fetchKuaishouIndustry } from './http.js'
|
||||
import { ensureKuaishouIndustryAccessToken } from './token-service.js'
|
||||
import { logIntegration } from '../../../utils/logger.js'
|
||||
import type { JsonObject } from '../../../types/json.js'
|
||||
@@ -108,8 +109,8 @@ export async function requestKuaishouIndustryOpenApi(
|
||||
const startedAt = Date.now()
|
||||
const res =
|
||||
httpMethod === 'GET'
|
||||
? await fetch(`${url}?${body.toString()}`, { method: 'GET' })
|
||||
: await fetch(url, {
|
||||
? await fetchKuaishouIndustry(`${url}?${body.toString()}`, { method: 'GET' })
|
||||
: await fetchKuaishouIndustry(url, {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/x-www-form-urlencoded' },
|
||||
body: body.toString(),
|
||||
|
||||
@@ -10,6 +10,7 @@ import {
|
||||
type KuaishouIndustryShopConfig,
|
||||
type KuaishouIndustrySourceConfig,
|
||||
} from './source-config-service.js'
|
||||
import { fetchKuaishouIndustry } from './http.js'
|
||||
|
||||
const ACCESS_TOKEN_REFRESH_MARGIN_MS = 30 * 60 * 1000
|
||||
const DEFAULT_ACCESS_TOKEN_TTL_MS = 47 * 60 * 60 * 1000
|
||||
@@ -314,8 +315,8 @@ async function requestKuaishouIndustryToken({
|
||||
const url = new URL(`${baseUrl}${path}`)
|
||||
const response =
|
||||
method === 'GET'
|
||||
? await fetch(appendSearchParams(url, params))
|
||||
: await fetch(url, {
|
||||
? await fetchKuaishouIndustry(appendSearchParams(url, params))
|
||||
: await fetchKuaishouIndustry(url, {
|
||||
method: 'POST',
|
||||
headers: { 'content-type': 'application/x-www-form-urlencoded' },
|
||||
body: params.toString(),
|
||||
|
||||
@@ -23,6 +23,23 @@ test('保留清理必须覆盖 work_order_events,否则这张只增不减的
|
||||
)
|
||||
})
|
||||
|
||||
test('保留任务要回填漏写的付款派生列,并在有命中时打日志', async () => {
|
||||
const src = await readJobSource()
|
||||
assert.ok(
|
||||
src.includes('source_order_amount_fen IS NULL'),
|
||||
'只处理派生列为空的行,避免把合法 0 或已落列的金额改掉',
|
||||
)
|
||||
assert.ok(
|
||||
src.includes("material_json->'source'->>'orderAmountFen'"),
|
||||
'回填口径必须与 083 / resolveSourceOrderAmountFen 相同',
|
||||
)
|
||||
assert.ok(src.includes("~ '^-?[0-9]{1,15}$'"), '超长数字不能 ::bigint,否则整次保留任务失败')
|
||||
assert.ok(
|
||||
src.includes('source_order_amount_fen 出现漏写'),
|
||||
'有命中必须打日志,否则金额静默回退 orders.total_amount',
|
||||
)
|
||||
})
|
||||
|
||||
test('保留清理默认窗口:敏感报文/事件 180 天,审计 365 天,工单事件 180 天', async () => {
|
||||
const defaults = await readFile(new URL('../../config/defaults.ts', import.meta.url), 'utf8')
|
||||
for (const line of [
|
||||
|
||||
@@ -1,5 +1,6 @@
|
||||
import { query } from '../../db/client.js'
|
||||
import { runtimeConfig } from '../../config/runtime.js'
|
||||
import { logWarn } from '../../utils/logger.js'
|
||||
import { getWorkerProductMatchConfig } from '../worker-platform/worker-product-match-config-service.js'
|
||||
import { pruneWorkProductMatchLogs } from '../../repositories/worker-platform/product-match-repo.js'
|
||||
|
||||
@@ -13,6 +14,7 @@ export type DataRetentionResult = {
|
||||
workOrderEventRows: number
|
||||
matchLogRows: number
|
||||
matchLogDeletedRows: number
|
||||
sourceOrderAmountBackfillRows: number
|
||||
}
|
||||
|
||||
/** 清理敏感原始报文并删除已过保留期的事件/审计记录。 */
|
||||
@@ -45,6 +47,14 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
|
||||
const webhookEventRows = await deleteOldRowsInBatches('webhook_events', webhookEventDays)
|
||||
const auditLogRows = await deleteOldRowsInBatches('admin_audit_logs', auditLogDays)
|
||||
const workOrderEventRows = await deleteOldRowsInBatches('work_order_events', workOrderEventDays)
|
||||
// 统计热路径已不再读 material_json。若将来有第三个写入点忘了维护派生列,
|
||||
// 金额会静默回退 orders.total_amount。这里把漏写补上并打日志,让口径变化可见。
|
||||
const sourceOrderAmountBackfillRows = await backfillMissingSourceOrderAmount()
|
||||
if (sourceOrderAmountBackfillRows > 0) {
|
||||
logWarn('[data-retention]', 'source_order_amount_fen 出现漏写,已按 JSON 回填', {
|
||||
rows: sourceOrderAmountBackfillRows,
|
||||
})
|
||||
}
|
||||
|
||||
return {
|
||||
status: 'ok',
|
||||
@@ -56,8 +66,23 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
|
||||
webhookEventRows,
|
||||
auditLogRows,
|
||||
workOrderEventRows,
|
||||
sourceOrderAmountBackfillRows,
|
||||
}
|
||||
}
|
||||
|
||||
/** 只补「列为 NULL 且 JSON 里仍有 1~15 位整数」的行。正常应为 0。 */
|
||||
async function backfillMissingSourceOrderAmount() {
|
||||
const result = await query(
|
||||
`
|
||||
UPDATE work_orders
|
||||
SET source_order_amount_fen = (material_json->'source'->>'orderAmountFen')::bigint
|
||||
WHERE source_order_amount_fen IS NULL
|
||||
AND COALESCE(material_json->'source'->>'orderAmountFen', '') ~ '^-?[0-9]{1,15}$'
|
||||
`,
|
||||
)
|
||||
return result.rowCount || 0
|
||||
}
|
||||
|
||||
async function scrubJsonColumnInBatches(
|
||||
table: 'orders' | 'kuaishou_industry_vouchers' | 'work_product_match_logs',
|
||||
dateColumn: 'created_at' | 'updated_at',
|
||||
|
||||
@@ -164,6 +164,7 @@ async function runJob(job: JsonObject, { manual = false }: { manual?: boolean }
|
||||
taskEventRows: Number(summary.taskEventRows || 0),
|
||||
webhookEventRows: Number(summary.webhookEventRows || 0),
|
||||
auditLogRows: Number(summary.auditLogRows || 0),
|
||||
sourceOrderAmountBackfillRows: Number(summary.sourceOrderAmountBackfillRows || 0),
|
||||
}
|
||||
: {}),
|
||||
})
|
||||
|
||||
@@ -27,15 +27,14 @@ import { getRequiredWorkOrder } from './worker-session-context-service.js'
|
||||
/**
|
||||
* 「本次筛选统计」的短缓存。
|
||||
*
|
||||
* 为什么要缓存:这条统计要扫全部命中工单(生产 9.6 万行宽表 + 关联 orders + 拼单汇总),
|
||||
* 实测稳定在 ~400ms。金额落到派生列后已经比原来(3.5~4.4 秒)快了约 9 倍,但**成本随工单量
|
||||
* 线性增长**——数据再翻几倍就会重新撞上 500ms 慢查询线。这里按"筛选指纹"缓存 10 秒并合并
|
||||
* 并发请求,把单位时间内的计算次数封顶在 TTL 上,而不是跟着访问量走。
|
||||
* 为什么要缓存:这条统计要扫全部命中工单(关联 orders + 拼单汇总)。派生列和覆盖索引
|
||||
* 把单次计算压到几十毫秒之后,10 秒 TTL 只会让分页总数滞后,体感就是「数量对不上」。
|
||||
* 这里按筛选指纹缓存 3 秒并合并并发请求:挡住同一瞬间的重复刷新,又把滞后收进一次轮询。
|
||||
*
|
||||
* 代价是统计与分页总数最多滞后 10 秒:期间操作产生的变化,前端已由实时事件快照就地修正
|
||||
* 期间操作产生的变化,前端已由实时事件快照就地修正
|
||||
* (见 admin-realtime-work-orders.ts 同步 statistics 的逻辑),刷新页面后也会自动对齐。
|
||||
*/
|
||||
const WORK_ORDER_STATISTICS_CACHE_TTL_MS = 10_000
|
||||
const WORK_ORDER_STATISTICS_CACHE_TTL_MS = 3_000
|
||||
const workOrderStatisticsCache = new WorkerListCache(WORK_ORDER_STATISTICS_CACHE_TTL_MS, 32)
|
||||
|
||||
/** 只把真正影响统计结果的筛选字段拼进指纹,避免无关参数造成缓存穿透。 */
|
||||
|
||||
@@ -2540,7 +2540,8 @@ test('筛选统计必须优先读派生列 source_order_amount_fen,不再逐
|
||||
new URL('../../repositories/worker-platform/work-order-query-repo.ts', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
// 统计表达式:列 → JSON 兜底 → orders.total_amount,COALESCE 从左到右短路
|
||||
// 统计表达式只读派生列。SQL 文本里一旦出现 material_json,规划器就会回表,
|
||||
// 执行器短路也救不了 index-only scan。
|
||||
assert.ok(
|
||||
queryRepo.includes('wo.source_order_amount_fen'),
|
||||
'统计没有走派生列,又会退化成每次解析全表 JSON(生产实测 2.8 秒)',
|
||||
@@ -2550,13 +2551,12 @@ test('筛选统计必须优先读派生列 source_order_amount_fen,不再逐
|
||||
queryRepo.indexOf('export async function getWorkOrderStatistics'),
|
||||
)
|
||||
assert.ok(
|
||||
paymentBlock.indexOf('wo.source_order_amount_fen') <
|
||||
paymentBlock.indexOf("material_json->'source'->>'orderAmountFen'"),
|
||||
'派生列必须排在 JSON 兜底之前,否则短路失效、等于没优化',
|
||||
!paymentBlock.includes('material_json'),
|
||||
'统计金额不能再引用 material_json,否则覆盖索引做不了 index-only scan',
|
||||
)
|
||||
assert.ok(
|
||||
paymentBlock.includes('COALESCE(o.total_amount, 0)'),
|
||||
'最后一级兜底必须与改造前一致(回退 orders.total_amount)',
|
||||
paymentBlock.includes('COALESCE(wo.source_order_amount_fen, o.total_amount, 0)'),
|
||||
'NULL 必须回退 orders.total_amount,与 083 列定义一致',
|
||||
)
|
||||
|
||||
// 两个写入点都要维护派生列,否则新单全是 NULL(虽然能兜底,但白优化)
|
||||
@@ -2604,9 +2604,10 @@ test('筛选统计必须有短缓存 + 并发合并,避免成本随工单量
|
||||
src.includes('workOrderStatisticsCache.getOrSet('),
|
||||
'统计必须走 getOrSet(同时合并并发请求,避免多人同时刷新时重复计算)',
|
||||
)
|
||||
assert.ok(
|
||||
/WORK_ORDER_STATISTICS_CACHE_TTL_MS = \d[\d_]*/.test(src),
|
||||
'TTL 必须是秒级常量:太长会让统计与分页总数明显滞后',
|
||||
assert.equal(
|
||||
src.includes('WORK_ORDER_STATISTICS_CACHE_TTL_MS = 3_000'),
|
||||
true,
|
||||
'统计降到几十毫秒后 TTL 保持 3 秒:再长会让分页总数滞后',
|
||||
)
|
||||
// 指纹必须覆盖所有影响统计的筛选字段,否则不同筛选会互相污染
|
||||
for (const field of [
|
||||
@@ -2685,6 +2686,57 @@ test('按运营态筛选的计数必须有覆盖索引(生产实测每次读 2
|
||||
)
|
||||
})
|
||||
|
||||
test('筛选统计覆盖索引必须吃掉 085,且不能为 VIP 把 JSON 列加回去', async () => {
|
||||
const { readFile } = await import('node:fs/promises')
|
||||
const migration = await readFile(
|
||||
new URL('../../db/migrations/087_work_order_statistics_cover_index.sql', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
assert.ok(migration.includes('idx_work_orders_statistics_cover'))
|
||||
for (const column of [
|
||||
'status',
|
||||
'sharing_enabled',
|
||||
'sharing_total_quantity',
|
||||
'source_order_amount_fen',
|
||||
'reward_amount',
|
||||
'order_id',
|
||||
]) {
|
||||
assert.ok(migration.includes(column), `统计覆盖索引缺少 ${column}`)
|
||||
}
|
||||
assert.ok(
|
||||
migration.includes('DROP INDEX IF EXISTS idx_work_orders_operational_status_cover'),
|
||||
'087 必须删掉 085,两条近乎重复的索引不能并存',
|
||||
)
|
||||
const workOrdersIndex = migration.slice(
|
||||
migration.indexOf('CREATE INDEX IF NOT EXISTS idx_work_orders_statistics_cover'),
|
||||
migration.indexOf('CREATE INDEX IF NOT EXISTS idx_orders_statistics_cover'),
|
||||
)
|
||||
assert.ok(!workOrdersIndex.includes('acceptance_json'), '不要把 acceptance_json 放进覆盖索引')
|
||||
assert.ok(!workOrdersIndex.includes('material_json'), '不要把 material_json 放进覆盖索引')
|
||||
assert.ok(
|
||||
migration.includes('idx_orders_statistics_cover'),
|
||||
'缺 orders 覆盖索引时,统计会把 26 万行 orders 整表扫来做 Hash Right Join',
|
||||
)
|
||||
assert.ok(migration.includes('INCLUDE (pay_status, order_status, total_amount)'))
|
||||
assert.ok(
|
||||
migration.includes('低于 5%'),
|
||||
'Heap Fetches 真空后仍有约 1.7%,判据不能写成必须为 0',
|
||||
)
|
||||
assert.ok(
|
||||
!/^\s*CREATE INDEX CONCURRENTLY/m.test(migration),
|
||||
'迁移在事务内执行,不能使用 CONCURRENTLY',
|
||||
)
|
||||
|
||||
const queryRepo = await readFile(
|
||||
new URL('../../repositories/worker-platform/work-order-query-repo.ts', import.meta.url),
|
||||
'utf8',
|
||||
)
|
||||
assert.ok(
|
||||
queryRepo.includes('不要把 acceptance_json 补进索引'),
|
||||
'VIP 回表降级必须写在统计查询旁,避免以后有人把 JSON 列加回索引',
|
||||
)
|
||||
})
|
||||
|
||||
test('三处"只差索引"的热查询缺口必须有迁移兜底(生产已前后对比验证)', async () => {
|
||||
const { readFile } = await import('node:fs/promises')
|
||||
const migration = await readFile(
|
||||
|
||||
@@ -1,30 +1,18 @@
|
||||
import { Tag } from 'antd'
|
||||
import { useEffect, useState } from 'react'
|
||||
|
||||
export function DeadlineCountdown({
|
||||
deadlineAt,
|
||||
showExpired: _showExpired = false,
|
||||
}: {
|
||||
deadlineAt: string
|
||||
showExpired?: boolean
|
||||
}) {
|
||||
import { resolveDeadlineCountdown } from '@/utils/deadline-countdown'
|
||||
|
||||
export function DeadlineCountdown({ deadlineAt }: { deadlineAt: string }) {
|
||||
const [now, setNow] = useState(() => Date.now())
|
||||
const view = resolveDeadlineCountdown(deadlineAt, now)
|
||||
|
||||
useEffect(() => {
|
||||
const timer = window.setInterval(() => setNow(Date.now()), 1000)
|
||||
const timer = window.setInterval(() => setNow(Date.now()), view.tickMs)
|
||||
return () => window.clearInterval(timer)
|
||||
}, [])
|
||||
}, [view.tickMs])
|
||||
|
||||
const remainingMs = new Date(deadlineAt).getTime() - now
|
||||
if (remainingMs <= 0) {
|
||||
return <Tag color="red">已超时</Tag>
|
||||
}
|
||||
const totalSeconds = Math.floor(remainingMs / 1000)
|
||||
const minutes = Math.floor(totalSeconds / 60)
|
||||
const seconds = totalSeconds % 60
|
||||
const label = minutes > 0 ? `${minutes} 分 ${seconds} 秒` : `${seconds} 秒`
|
||||
const color = remainingMs < 10 * 60 * 1000 ? 'orange' : 'blue'
|
||||
return <Tag color={color}>剩余 {label}</Tag>
|
||||
return <Tag color={view.color}>{view.label}</Tag>
|
||||
}
|
||||
|
||||
export function formatTimeoutPolicyLabel(policy?: string) {
|
||||
|
||||
@@ -1472,7 +1472,7 @@ export default function WorkOrdersPanel() {
|
||||
</Typography.Text>
|
||||
</Space>
|
||||
{row.status === 'in_progress' && row.deadlineAt ? (
|
||||
<DeadlineCountdown deadlineAt={row.deadlineAt} showExpired />
|
||||
<DeadlineCountdown deadlineAt={row.deadlineAt} />
|
||||
) : null}
|
||||
{row.deadlineAt ? (
|
||||
<Typography.Text type="secondary">
|
||||
@@ -2103,7 +2103,7 @@ export default function WorkOrdersPanel() {
|
||||
<Descriptions.Item label="超时截止">
|
||||
{detailOrder.deadlineAt ? (
|
||||
<Space wrap size={8}>
|
||||
<DeadlineCountdown deadlineAt={detailOrder.deadlineAt} showExpired />
|
||||
<DeadlineCountdown deadlineAt={detailOrder.deadlineAt} />
|
||||
<Typography.Text type="secondary">
|
||||
{formatAdminDateTime(detailOrder.deadlineAt)}
|
||||
</Typography.Text>
|
||||
|
||||
@@ -141,13 +141,12 @@ export default function WorkerHallPage() {
|
||||
})
|
||||
|
||||
const hallSummaryQuery = useQuery({
|
||||
queryKey: ['worker-hall-summary'],
|
||||
queryFn: async () => {
|
||||
const response = await fetchWorkerMyOrderOverview()
|
||||
queryKey: ['worker-my-orders-overview'],
|
||||
queryFn: fetchWorkerMyOrderOverview,
|
||||
select: (response) => {
|
||||
const summaryByStatus = new Map(
|
||||
response.data.items.map((item) => [item.status, item.orderCount]),
|
||||
)
|
||||
|
||||
return {
|
||||
pendingAcceptanceCount: Number(summaryByStatus.get('pending_acceptance') || 0),
|
||||
problemCount: Number(summaryByStatus.get('problem') || 0),
|
||||
@@ -283,7 +282,6 @@ export default function WorkerHallPage() {
|
||||
await Promise.all([
|
||||
ordersQuery.refetch(),
|
||||
queryClient.invalidateQueries({ queryKey: ['worker-hall-categories'] }),
|
||||
queryClient.invalidateQueries({ queryKey: ['worker-hall-summary'] }),
|
||||
queryClient.invalidateQueries({ queryKey: ['worker-profile'] }),
|
||||
queryClient.invalidateQueries({ queryKey: ['worker-my-orders'] }),
|
||||
queryClient.invalidateQueries({ queryKey: ['worker-my-orders-overview'] }),
|
||||
|
||||
@@ -206,11 +206,13 @@ export default function WorkerOrdersPage() {
|
||||
const cancelRequestsQuery = useQuery({
|
||||
queryKey: ['worker-cancel-requests'],
|
||||
queryFn: fetchWorkerCancelRequests,
|
||||
enabled: Boolean(cancelRequestOrder),
|
||||
retry: false,
|
||||
})
|
||||
const feedbacksQuery = useQuery({
|
||||
queryKey: ['worker-order-feedbacks'],
|
||||
queryFn: fetchWorkerOrderFeedbacks,
|
||||
enabled: Boolean(feedbackOrder),
|
||||
retry: false,
|
||||
})
|
||||
const overviewByStatus = new Map(
|
||||
|
||||
@@ -0,0 +1,30 @@
|
||||
export const DEADLINE_FAST_TICK_MS = 1_000
|
||||
export const DEADLINE_SLOW_TICK_MS = 30_000
|
||||
export const DEADLINE_FAST_WINDOW_MS = 60_000
|
||||
|
||||
export function resolveDeadlineCountdown(deadlineAt: string, now: number) {
|
||||
const remainingMs = new Date(deadlineAt).getTime() - now
|
||||
if (!Number.isFinite(remainingMs) || remainingMs <= 0) {
|
||||
return {
|
||||
label: '已超时',
|
||||
color: 'red' as const,
|
||||
tickMs: DEADLINE_SLOW_TICK_MS,
|
||||
}
|
||||
}
|
||||
|
||||
const totalSeconds = Math.floor(remainingMs / 1000)
|
||||
const minutes = Math.floor(totalSeconds / 60)
|
||||
const seconds = totalSeconds % 60
|
||||
const fast = remainingMs < DEADLINE_FAST_WINDOW_MS
|
||||
const label = fast
|
||||
? minutes > 0
|
||||
? `剩余 ${minutes} 分 ${seconds} 秒`
|
||||
: `剩余 ${seconds} 秒`
|
||||
: `剩余 ${minutes} 分`
|
||||
|
||||
return {
|
||||
label,
|
||||
color: (remainingMs < 10 * 60_000 ? 'orange' : 'blue') as 'orange' | 'blue',
|
||||
tickMs: fast ? DEADLINE_FAST_TICK_MS : DEADLINE_SLOW_TICK_MS,
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,35 @@
|
||||
import assert from 'node:assert/strict'
|
||||
import test from 'node:test'
|
||||
|
||||
import {
|
||||
DEADLINE_FAST_TICK_MS,
|
||||
DEADLINE_SLOW_TICK_MS,
|
||||
resolveDeadlineCountdown,
|
||||
} from '../src/utils/deadline-countdown.ts'
|
||||
|
||||
const now = Date.parse('2026-10-01T06:00:00.000Z')
|
||||
|
||||
test('deadline under one minute ticks every second and keeps the seconds label', () => {
|
||||
const view = resolveDeadlineCountdown('2026-10-01T06:00:45.000Z', now)
|
||||
assert.equal(view.tickMs, DEADLINE_FAST_TICK_MS)
|
||||
assert.equal(view.label, '剩余 45 秒')
|
||||
assert.equal(view.color, 'orange')
|
||||
})
|
||||
|
||||
test('deadline over one minute ticks every 30 seconds and hides seconds', () => {
|
||||
const near = resolveDeadlineCountdown('2026-10-01T06:02:20.000Z', now)
|
||||
assert.equal(near.tickMs, DEADLINE_SLOW_TICK_MS)
|
||||
assert.equal(near.label, '剩余 2 分')
|
||||
assert.equal(near.color, 'orange')
|
||||
|
||||
const far = resolveDeadlineCountdown('2026-10-01T06:12:20.000Z', now)
|
||||
assert.equal(far.label, '剩余 12 分')
|
||||
assert.equal(far.color, 'blue')
|
||||
})
|
||||
|
||||
test('expired deadline stops the one-second timer', () => {
|
||||
const view = resolveDeadlineCountdown('2026-10-01T05:59:00.000Z', now)
|
||||
assert.equal(view.label, '已超时')
|
||||
assert.equal(view.color, 'red')
|
||||
assert.equal(view.tickMs, DEADLINE_SLOW_TICK_MS)
|
||||
})
|
||||
Reference in New Issue
Block a user