merge: 合并 main 到 dev(含订单来源快照与寄售改名同步)

# Conflicts:
#	README.md
#	apps/backend/src/db/migrate.ts
This commit is contained in:
2026-11-12 15:35:42 +08:00
38 changed files with 860 additions and 289 deletions
-6
View File
@@ -72,12 +72,6 @@ LOG_RETENTION_DAYS=30
LOG_MAX_FILE_SIZE_MB=20 LOG_MAX_FILE_SIZE_MB=20
LOG_MAX_FILES=5 LOG_MAX_FILES=5
# 数据库敏感报文/事件保留策略(天)。清理任务每天执行一次。
RAW_PAYLOAD_RETENTION_DAYS=180
TASK_EVENT_RETENTION_DAYS=180
WEBHOOK_EVENT_RETENTION_DAYS=180
AUDIT_LOG_RETENTION_DAYS=365
WORK_ORDER_EVENT_RETENTION_DAYS=180
# 已处理成本异常保留天数,超期搬入 cost_exceptions_archive(open 行不归档) # 已处理成本异常保留天数,超期搬入 cost_exceptions_archive(open 行不归档)
COST_EXCEPTION_RETENTION_DAYS=60 COST_EXCEPTION_RETENTION_DAYS=60
-6
View File
@@ -24,12 +24,6 @@ export function createDefaultRuntimeConfig(projectRoot: string): RuntimeConfig {
}, },
retention: { retention: {
// 原始报文只保留追溯所需窗口,清理后保留订单/事件摘要。
rawPayloadDays: 180,
taskEventDays: 180,
webhookEventDays: 180,
auditLogDays: 365,
workOrderEventDays: 180,
// 已处理的成本异常保留 60 天,之后搬进 cost_exceptions_archive;open 行不归档。 // 已处理的成本异常保留 60 天,之后搬进 cost_exceptions_archive;open 行不归档。
costExceptionDays: 60, costExceptionDays: 60,
}, },
@@ -96,10 +96,6 @@ function createRuntimeConfig(): RuntimeConfig {
retentionDays: 30, retentionDays: 30,
}, },
retention: { retention: {
rawPayloadDays: 180,
taskEventDays: 180,
webhookEventDays: 180,
auditLogDays: 365,
costExceptionDays: 60, costExceptionDays: 60,
}, },
database: { database: {
-5
View File
@@ -30,11 +30,6 @@ export const ENV_OVERRIDES: readonly EnvOverride[] = [
integerEnv('LOG_RETENTION_DAYS', ['logging', 'retentionDays']), integerEnv('LOG_RETENTION_DAYS', ['logging', 'retentionDays']),
integerEnv('LOG_MAX_FILE_SIZE_MB', ['logging', 'maxFileSizeMb']), integerEnv('LOG_MAX_FILE_SIZE_MB', ['logging', 'maxFileSizeMb']),
integerEnv('LOG_MAX_FILES', ['logging', 'maxFiles']), integerEnv('LOG_MAX_FILES', ['logging', 'maxFiles']),
integerEnv('RAW_PAYLOAD_RETENTION_DAYS', ['retention', 'rawPayloadDays']),
integerEnv('TASK_EVENT_RETENTION_DAYS', ['retention', 'taskEventDays']),
integerEnv('WEBHOOK_EVENT_RETENTION_DAYS', ['retention', 'webhookEventDays']),
integerEnv('AUDIT_LOG_RETENTION_DAYS', ['retention', 'auditLogDays']),
integerEnv('WORK_ORDER_EVENT_RETENTION_DAYS', ['retention', 'workOrderEventDays']),
integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays']), integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays']),
stringEnv('DATABASE_URL', ['database', 'url']), stringEnv('DATABASE_URL', ['database', 'url']),
booleanEnv('DATABASE_SSL', ['database', 'ssl']), booleanEnv('DATABASE_SSL', ['database', 'ssl']),
@@ -160,10 +160,6 @@ function createRuntimeConfig(overrides: Record<string, unknown> = {}): RuntimeCo
integrationLevel: 'warn', integrationLevel: 'warn',
}, },
retention: { retention: {
rawPayloadDays: 180,
taskEventDays: 180,
webhookEventDays: 180,
auditLogDays: 365,
costExceptionDays: 60, costExceptionDays: 60,
}, },
database: { database: {
@@ -82,12 +82,6 @@ export function validateRuntimeConfig(
max: 100, max: 100,
}) })
if (config.retention) { if (config.retention) {
requireInteger(issues, 'retention.rawPayloadDays', config.retention.rawPayloadDays, { min: 1 })
requireInteger(issues, 'retention.taskEventDays', config.retention.taskEventDays, { min: 1 })
requireInteger(issues, 'retention.webhookEventDays', config.retention.webhookEventDays, {
min: 1,
})
requireInteger(issues, 'retention.auditLogDays', config.retention.auditLogDays, { min: 1 })
requireInteger(issues, 'retention.costExceptionDays', config.retention.costExceptionDays, { requireInteger(issues, 'retention.costExceptionDays', config.retention.costExceptionDays, {
min: 1, min: 1,
}) })
+5 -8
View File
@@ -64,13 +64,6 @@ export async function runDatabaseMigrations() {
console.info(`[db:migrate] 迁移目录: ${MIGRATIONS_DIR}`) console.info(`[db:migrate] 迁移目录: ${MIGRATIONS_DIR}`)
// 分支联调库可能已应用过 main 的 147-158,而本分支的寄卖迁移编号是 147-152;
// node-pg-migrate 的 checkOrder 会按名称逐位比较并直接抛错,导致服务起不来。
// 生产保持严格顺序;开发/测试允许顺序放宽(迁移文件本身仍需幂等)。
const strictMigrationOrder =
String(process.env.DB_MIGRATION_STRICT_ORDER || '').trim() === '1' ||
String(process.env.NODE_ENV || '').trim() === 'production'
await runner({ await runner({
databaseUrl: createMigrationDatabaseConfig(), databaseUrl: createMigrationDatabaseConfig(),
dir: MIGRATIONS_DIR, dir: MIGRATIONS_DIR,
@@ -79,7 +72,11 @@ export async function runDatabaseMigrations() {
migrationsSchema: 'public', migrationsSchema: 'public',
schema: 'public', schema: 'public',
singleTransaction: true, singleTransaction: true,
checkOrder: strictMigrationOrder, // 联调/生产库可能存在「已应用、但当前分支已删除」的历史迁移记录
// (如 153-155 semi_auto_*)。node-pg-migrate 的严格顺序检查会把这类
// 残留记录判成乱序并直接拒绝启动;本地迁移文件本身只增不改,
// 这里关闭顺序检查,仍会按文件名顺序执行所有未应用迁移。
checkOrder: false,
logger: { logger: {
debug: () => undefined, debug: () => undefined,
info: (message) => { info: (message) => {
@@ -0,0 +1,57 @@
-- 152_order_source_facts_immutable.sql
--
-- 永久保留平台订单事实:
-- 1. 每次来源同步追加完整订单快照,供审计和后续重放;
-- 2. 快照只允许插入,不允许应用更新或删除;
-- 3. orders.raw_payload_json 由应用写入口保证首次写入后不再覆盖。
CREATE TABLE IF NOT EXISTS order_source_snapshots (
id BIGSERIAL PRIMARY KEY,
order_id BIGINT NOT NULL REFERENCES orders(id) ON DELETE RESTRICT,
provider TEXT NOT NULL,
platform TEXT NOT NULL,
shop_id TEXT NOT NULL DEFAULT '',
platform_order_id TEXT NOT NULL,
source_label TEXT NOT NULL DEFAULT '',
payload_json JSONB NOT NULL,
payload_sha256 TEXT NOT NULL,
recorded_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
CONSTRAINT chk_order_source_snapshots_payload_sha256
CHECK (payload_sha256 ~ '^[0-9a-f]{64}$'),
CONSTRAINT uq_order_source_snapshots_payload
UNIQUE (order_id, payload_sha256)
);
CREATE INDEX IF NOT EXISTS idx_order_source_snapshots_order
ON order_source_snapshots(order_id, recorded_at DESC, id DESC);
CREATE INDEX IF NOT EXISTS idx_order_source_snapshots_platform_order
ON order_source_snapshots(provider, platform, platform_order_id, recorded_at DESC);
COMMENT ON TABLE order_source_snapshots IS
'平台订单同步的不可变原始快照;重复同步只追加,不覆盖订单主表原始报文';
COMMENT ON COLUMN order_source_snapshots.payload_json IS
'平台来源事件、原始报文与订单项完整快照';
COMMENT ON COLUMN order_source_snapshots.payload_sha256 IS
'完整快照内容 SHA-256,用于同一订单重复消息幂等';
CREATE OR REPLACE FUNCTION reject_order_source_snapshot_mutation()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
IF current_setting('app.allow_order_fact_repair', TRUE) = 'on' THEN
RETURN CASE WHEN TG_OP = 'DELETE' THEN OLD ELSE NEW END;
END IF;
RAISE EXCEPTION 'order_source_snapshots is append-only'
USING ERRCODE = '23514';
END;
$$;
DROP TRIGGER IF EXISTS trg_order_source_snapshots_append_only
ON order_source_snapshots;
CREATE TRIGGER trg_order_source_snapshots_append_only
BEFORE UPDATE OR DELETE ON order_source_snapshots
FOR EACH ROW
EXECUTE FUNCTION reject_order_source_snapshot_mutation();
@@ -0,0 +1,75 @@
-- 158_consignment_worker_name_sync.sql —— 打手改名后同步寄售登记人快照。
--
-- 背景:consignment_inventories.worker_name / consignment_submissions.worker_name
-- 是入库/提报时的名字快照。worker_users.display_name 改名后不会自动更新,
-- 后台寄售列表仍显示旧名,按新名搜索也搜不到。
--
-- 这里做两件事:
-- 1. 用当前昵称/电话回填存量快照;
-- 2. worker_users 改名(display_name/username/phone)时自动同步这两张表,
-- 之后新改名无需逐表手工处理。
--
-- 历史工单 material_json.consignment.workerName 不做批量改写:
-- 后台展示端按 consignment_inventories 的实时关联覆盖,保留历史 JSON 的登记痕迹。
UPDATE consignment_inventories ci
SET
worker_name = COALESCE(NULLIF(wu.display_name, ''), wu.username),
worker_phone = wu.phone
FROM worker_users wu
WHERE ci.worker_id = wu.id
AND (
ci.worker_name IS DISTINCT FROM COALESCE(NULLIF(wu.display_name, ''), wu.username)
OR ci.worker_phone IS DISTINCT FROM wu.phone
);
UPDATE consignment_submissions cs
SET
worker_name = COALESCE(NULLIF(wu.display_name, ''), wu.username),
worker_phone = wu.phone
FROM worker_users wu
WHERE cs.worker_id = wu.id
AND (
cs.worker_name IS DISTINCT FROM COALESCE(NULLIF(wu.display_name, ''), wu.username)
OR cs.worker_phone IS DISTINCT FROM wu.phone
);
CREATE OR REPLACE FUNCTION sync_worker_name_to_consignment_snapshots()
RETURNS TRIGGER AS $$
DECLARE
next_worker_name TEXT := COALESCE(NULLIF(NEW.display_name, ''), NEW.username);
BEGIN
IF NEW.display_name IS DISTINCT FROM OLD.display_name
OR NEW.username IS DISTINCT FROM OLD.username
OR NEW.phone IS DISTINCT FROM OLD.phone
THEN
UPDATE consignment_inventories
SET
worker_name = next_worker_name,
worker_phone = NEW.phone
WHERE worker_id = NEW.id
AND (
worker_name IS DISTINCT FROM next_worker_name
OR worker_phone IS DISTINCT FROM NEW.phone
);
UPDATE consignment_submissions
SET
worker_name = next_worker_name,
worker_phone = NEW.phone
WHERE worker_id = NEW.id
AND (
worker_name IS DISTINCT FROM next_worker_name
OR worker_phone IS DISTINCT FROM NEW.phone
);
END IF;
RETURN NEW;
END;
$$ LANGUAGE plpgsql;
DROP TRIGGER IF EXISTS trg_worker_users_consignment_name_sync ON worker_users;
CREATE TRIGGER trg_worker_users_consignment_name_sync
AFTER UPDATE OF display_name, username, phone ON worker_users
FOR EACH ROW
EXECUTE FUNCTION sync_worker_name_to_consignment_snapshots();
@@ -38,7 +38,7 @@ test('resolveOrderItemSyncPlan preserves matching order item ids across repeated
}, },
]) ])
assert.deepEqual(plan.creates, []) assert.deepEqual(plan.creates, [])
assert.deepEqual(plan.deletes, []) assert.deepEqual(plan.surplusExistingIds, [])
}) })
test('resolveOrderItemSyncPlan matches by identity before falling back to position', () => { test('resolveOrderItemSyncPlan matches by identity before falling back to position', () => {
@@ -74,10 +74,10 @@ test('resolveOrderItemSyncPlan matches by identity before falling back to positi
[22, 21], [22, 21],
) )
assert.deepEqual(plan.creates, []) assert.deepEqual(plan.creates, [])
assert.deepEqual(plan.deletes, []) assert.deepEqual(plan.surplusExistingIds, [])
}) })
test('resolveOrderItemSyncPlan keeps surplus existing ids for conditional cleanup instead of recreating everything', () => { test('resolveOrderItemSyncPlan reports surplus existing items without deleting them', () => {
const existingItems = [ const existingItems = [
{ id: 31, sku_code: 'sku-a', sku_name: '礼包 A', quantity: 1 }, { id: 31, sku_code: 'sku-a', sku_name: '礼包 A', quantity: 1 },
{ id: 32, sku_code: 'sku-b', sku_name: '礼包 B', quantity: 1 }, { id: 32, sku_code: 'sku-b', sku_name: '礼包 B', quantity: 1 },
@@ -101,7 +101,7 @@ test('resolveOrderItemSyncPlan keeps surplus existing ids for conditional cleanu
[31], [31],
) )
assert.deepEqual(plan.creates, []) assert.deepEqual(plan.creates, [])
assert.deepEqual(plan.deletes, [32]) assert.deepEqual(plan.surplusExistingIds, [32])
}) })
test('已分配订单重推时冻结原渠道决策与中文内部 SKU', () => { test('已分配订单重推时冻结原渠道决策与中文内部 SKU', () => {
@@ -155,7 +155,7 @@ test('已分配订单重推时冻结原渠道决策与中文内部 SKU', () => {
}) })
}) })
test('已分配订单重推写库时不会用平台新 SKU 覆盖冻结值', async () => { test('平台订单重推只追加新订单项,不更新或删除已有原始订单项', async () => {
const existing = { const existing = {
id: '41', id: '41',
order_id: 7, order_id: 7,
@@ -184,24 +184,37 @@ test('已分配订单重推写库时不会用平台新 SKU 覆盖冻结值', asy
} }
const updateParams: unknown[][] = [] const updateParams: unknown[][] = []
let selectCount = 0 let selectCount = 0
let writeSql = ''
const executor = async (sql: string, params: unknown[] = []) => { const executor = async (sql: string, params: unknown[] = []) => {
if (sql.includes('SELECT *') && sql.includes('FROM v_order_item_current')) { if (sql.includes('SELECT *') && sql.includes('FROM v_order_item_current')) {
selectCount += 1 selectCount += 1
return { rows: selectCount === 1 ? [existing] : [] } return { rows: selectCount === 1 ? [existing] : [{ ...existing, id: 42 }] }
} }
if (sql.includes('UPDATE order_items')) { if (sql.includes('INSERT INTO order_items')) {
updateParams.push(params) updateParams.push(params)
writeSql += sql
return { rows: [], rowCount: 1 } return { rows: [], rowCount: 1 }
} }
if (/UPDATE\s+order_items|DELETE\s+FROM\s+order_items/i.test(sql)) {
throw new Error(`permanent source item must not be mutated: ${sql}`)
}
throw new Error(`unexpected SQL: ${sql}`) throw new Error(`unexpected SQL: ${sql}`)
} }
await replaceOrderItemsWithExecutor(executor as never, 7, [incoming]) await replaceOrderItemsWithExecutor(executor as never, 7, [
incoming,
{
...incoming,
skuCode: 'platform-extra',
skuName: '平台新增商品',
},
])
assert.equal(updateParams.length, 1) assert.equal(updateParams.length, 1)
assert.equal(updateParams[0]?.[0], 'lucky_coin_x2') assert.ok(writeSql.includes('INSERT INTO order_items'))
assert.equal(updateParams[0]?.[1], '幸运币2个') assert.equal(updateParams[0]?.[1], 'platform-extra')
assert.match(String(updateParams[0]?.[2]), /平台新增商品/)
}) })
test('所有替换订单项入口都在同一事务内分摊订单收入', async () => { test('所有替换订单项入口都在同一事务内分摊订单收入', async () => {
@@ -13,7 +13,7 @@ type OrderItemSyncPlan = {
item: OrderItemReplaceInput item: OrderItemReplaceInput
}> }>
creates: OrderItemReplaceInput[] creates: OrderItemReplaceInput[]
deletes: number[] surplusExistingIds: number[]
} }
type OrderItemIdentityInput = Partial<OrderItemReplaceInput> & Partial<OrderItemRow> type OrderItemIdentityInput = Partial<OrderItemReplaceInput> & Partial<OrderItemRow>
@@ -100,36 +100,6 @@ export async function replaceOrderItemsWithExecutor(
): Promise<OrderItemRow[]> { ): Promise<OrderItemRow[]> {
const existingItems = await listOrderItemsByOrderIdWithExecutor(executor, orderId) const existingItems = await listOrderItemsByOrderIdWithExecutor(executor, orderId)
const plan = resolveOrderItemSyncPlan(existingItems, items) const plan = resolveOrderItemSyncPlan(existingItems, items)
const existingById = new Map(existingItems.map((item) => [Number(item.id), item]))
for (const update of plan.updates) {
const effectiveItem = preserveAssignedOrderItemState(
existingById.get(Number(update.orderItemId)),
update.item,
)
await executor(
`
UPDATE order_items
SET
sku_code = $1,
sku_name = $2,
quantity = $3,
spec_json = $4::jsonb,
item_snapshot_json = $5::jsonb,
updated_at = $6
WHERE id = $7
`,
[
effectiveItem.skuCode,
effectiveItem.skuName,
update.item.quantity,
safeJsonStringify(effectiveItem.specJson || '{}'),
safeJsonStringify(effectiveItem.itemSnapshotJson || effectiveItem.specJson || '{}'),
effectiveItem.updatedAt,
update.orderItemId,
],
)
}
for (const create of plan.creates) { for (const create of plan.creates) {
await executor( await executor(
@@ -158,14 +128,6 @@ export async function replaceOrderItemsWithExecutor(
) )
} }
if (plan.deletes.length > 0) {
const deletableIds = await listDeletableOrderItemIdsWithExecutor(executor, plan.deletes)
if (deletableIds.length > 0) {
await executor('DELETE FROM order_items WHERE id = ANY($1::bigint[])', [deletableIds])
}
}
return listOrderItemsByOrderIdWithExecutor(executor, orderId) return listOrderItemsByOrderIdWithExecutor(executor, orderId)
} }
@@ -280,7 +242,9 @@ export function resolveOrderItemSyncPlan(
return { return {
updates, updates,
creates, creates,
deletes: unmatchedExisting.map((item) => Number(item.id)).filter((value) => value > 0), surplusExistingIds: unmatchedExisting
.map((item) => Number(item.id))
.filter((value) => value > 0),
} }
} }
@@ -301,34 +265,6 @@ function buildOrderItemIdentity(item: OrderItemIdentityInput): string {
].join('::') ].join('::')
} }
async function listDeletableOrderItemIdsWithExecutor(
executor: QueryExecutor,
orderItemIds: number[],
): Promise<number[]> {
const normalizedIds = Array.isArray(orderItemIds)
? orderItemIds.map((value) => Number(value)).filter((value) => value > 0)
: []
if (normalizedIds.length === 0) {
return []
}
const result = await executor(
`
SELECT oi.id
FROM v_order_item_current oi
LEFT JOIN fulfillment_tasks ft ON ft.order_item_id = oi.id
WHERE oi.id = ANY($1::bigint[])
GROUP BY oi.id
HAVING COUNT(ft.id) = 0
ORDER BY oi.id ASC
`,
[normalizedIds],
)
return result.rows.map((row) => Number(row.id)).filter((value) => value > 0)
}
async function listOrderItemsByOrderIdWithExecutor( async function listOrderItemsByOrderIdWithExecutor(
executor: QueryExecutor, executor: QueryExecutor,
orderId: number | string, orderId: number | string,
@@ -1,8 +1,18 @@
import assert from 'node:assert/strict' import assert from 'node:assert/strict'
import { readFile } from 'node:fs/promises'
import test from 'node:test' import test from 'node:test'
import { listOrdersWithExecutor, type OrderQueryExecutor } from './order-repo.js' import { listOrdersWithExecutor, type OrderQueryExecutor } from './order-repo.js'
test('订单更新永不覆盖首次写入的平台原始报文', async () => {
const source = await readFile(new URL('./order-repo.ts', import.meta.url), 'utf8')
const updateSource = source.slice(
source.indexOf('export async function updateOrderWithExecutor'),
source.indexOf('export async function getOrderById'),
)
assert.doesNotMatch(updateSource, /raw_payload_json\s*=/)
})
test('订单列表综合筛选下推到总数和分页查询', async () => { test('订单列表综合筛选下推到总数和分页查询', async () => {
const calls: Array<{ sql: string; params: unknown[] }> = [] const calls: Array<{ sql: string; params: unknown[] }> = []
const executor = createExecutor(calls, 3) const executor = createExecutor(calls, 3)
+3 -5
View File
@@ -191,10 +191,9 @@ export async function updateOrderWithExecutor(
receiver_contact = $9, receiver_contact = $9,
total_amount = $10, total_amount = $10,
currency = $11, currency = $11,
raw_payload_json = $12::jsonb, paid_at = $12,
paid_at = $13, updated_at = $13
updated_at = $14 WHERE id = $14
WHERE id = $15
RETURNING * RETURNING *
`, `,
[ [
@@ -209,7 +208,6 @@ export async function updateOrderWithExecutor(
sanitizeWellFormedUnicode(input.receiverContact), sanitizeWellFormedUnicode(input.receiverContact),
input.totalAmount, input.totalAmount,
input.currency, input.currency,
safeJsonStringify(input.rawPayloadJson || '{}'),
input.paidAt || null, input.paidAt || null,
input.updatedAt, input.updatedAt,
Number(orderId), Number(orderId),
@@ -0,0 +1,137 @@
/**
* 平台订单原始事实不可变集成测试(需要真实 PostgreSQL)。
*
* ORDER_SOURCE_FACTS_TEST_DATABASE_URL=postgres://... \
* npx tsx --test src/repositories/order-source-facts.integration.test.ts
*/
import assert from 'node:assert/strict'
import test from 'node:test'
const TEST_DATABASE_URL = String(process.env.ORDER_SOURCE_FACTS_TEST_DATABASE_URL || '').trim()
test(
'平台重推只追加原始事实,不覆盖订单报文、不删除已有订单项',
{ skip: !TEST_DATABASE_URL ? '未设置 ORDER_SOURCE_FACTS_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query, withTransaction } = await import('../db/client.js')
const { appendOrderSourceSnapshotWithExecutor } =
await import('./order-source-snapshot-repo.js')
const { replaceOrderItemsWithExecutor } = await import('./order-item-repo.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
let orderId = 0
try {
const order = await query<{ id: number }>(
`
INSERT INTO orders (
provider, platform, shop_id, shop_name, platform_order_id,
order_status, pay_status, total_amount, raw_payload_json, created_at, updated_at
) VALUES (
'91kaquan', 'kuaishou-91', $1, '原始事实测试店', $2,
'paid', 'paid', 1000, $3::jsonb, NOW(), NOW()
) RETURNING id
`,
[
`source-facts-shop-${suffix}`,
`SOURCE-FACTS-${suffix}`,
JSON.stringify({ orderNo: `SOURCE-FACTS-${suffix}`, version: 1 }),
],
)
orderId = Number(order.rows[0]?.id || 0)
assert.ok(orderId > 0)
const item = await query<{ id: number }>(
`
INSERT INTO order_items (
order_id, sku_code, sku_name, quantity, spec_json, item_snapshot_json,
revenue_amount_fen, revenue_basis, created_at, updated_at
) VALUES (
$1, 'sku-old', '原始商品', 1, '{}'::jsonb, '{}'::jsonb,
1000, 'order_total', NOW(), NOW()
) RETURNING id
`,
[orderId],
)
const originalItemId = Number(item.rows[0]?.id || 0)
assert.ok(originalItemId > 0)
const snapshotInput = {
orderId,
provider: '91kaquan',
platform: 'kuaishou-91',
shopId: `source-facts-shop-${suffix}`,
platformOrderId: `SOURCE-FACTS-${suffix}`,
sourceLabel: 'integration-test',
payload: { orderNo: `SOURCE-FACTS-${suffix}`, version: 1, items: ['sku-old'] },
recordedAt: new Date().toISOString(),
}
assert.equal(await appendOrderSourceSnapshotWithExecutor(query, snapshotInput), true)
assert.equal(await appendOrderSourceSnapshotWithExecutor(query, snapshotInput), false)
const snapshots = await query<{ count: number }>(
'SELECT COUNT(*)::int AS count FROM order_source_snapshots WHERE order_id = $1',
[orderId],
)
assert.equal(Number(snapshots.rows[0]?.count), 1)
await replaceOrderItemsWithExecutor(query, orderId, [
{
skuCode: 'sku-new',
skuName: '平台新商品',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
{
skuCode: 'sku-extra',
skuName: '平台新增商品',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
])
const items = await query<{ original_exists: boolean; total: number }>(
`
SELECT
EXISTS (SELECT 1 FROM order_items WHERE id = $1) AS original_exists,
(SELECT COUNT(*)::int FROM order_items WHERE order_id = $2) AS total
`,
[originalItemId, orderId],
)
assert.equal(items.rows[0]?.original_exists, true)
assert.equal(Number(items.rows[0]?.total), 2)
await replaceOrderItemsWithExecutor(query, orderId, [
{
skuCode: 'sku-new',
skuName: '平台新商品',
quantity: 1,
specJson: '{}',
itemSnapshotJson: '{}',
createdAt: new Date().toISOString(),
updatedAt: new Date().toISOString(),
},
])
const afterShrink = await query<{ total: number }>(
'SELECT COUNT(*)::int AS total FROM order_items WHERE order_id = $1',
[orderId],
)
assert.equal(Number(afterShrink.rows[0]?.total), 2)
} finally {
if (orderId > 0) {
await withTransaction(async (client) => {
await client.query(`SET LOCAL app.allow_order_fact_repair = 'on'`)
await client.query('DELETE FROM order_source_snapshots WHERE order_id = $1', [orderId])
await client.query('DELETE FROM orders WHERE id = $1', [orderId])
})
}
await closeDb()
}
},
)
@@ -0,0 +1,40 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import {
appendOrderSourceSnapshotWithExecutor,
orderSourceSnapshotHash,
} from './order-source-snapshot-repo.js'
test('订单来源快照哈希稳定且能区分不同报文', () => {
const first = orderSourceSnapshotHash({ orderNo: 'A-1', status: 'paid' })
assert.equal(first, orderSourceSnapshotHash({ orderNo: 'A-1', status: 'paid' }))
assert.notEqual(first, orderSourceSnapshotHash({ orderNo: 'A-1', status: 'refunded' }))
assert.match(first, /^[0-9a-f]{64}$/)
})
test('同订单相同来源快照只追加一次', async () => {
const calls: unknown[][] = []
const executor = async (sql: string, params: unknown[] = []) => {
calls.push(params)
assert.ok(sql.includes('INSERT INTO order_source_snapshots'))
assert.ok(sql.includes('ON CONFLICT (order_id, payload_sha256) DO NOTHING'))
return { rows: [{ id: 1 }] }
}
const inserted = await appendOrderSourceSnapshotWithExecutor(executor as never, {
orderId: 7,
provider: 'kuaishou-91',
platform: 'kuaishou-91',
shopId: 'shop-1',
platformOrderId: 'ORDER-1',
sourceLabel: 'source-order-upsert',
payload: { orderNo: 'ORDER-1', status: 'paid' },
recordedAt: '2026-11-09T00:00:00.000Z',
})
assert.equal(inserted, true)
assert.equal(calls.length, 1)
assert.equal(calls[0]?.[0], 7)
assert.equal(calls[0]?.[7], orderSourceSnapshotHash({ orderNo: 'ORDER-1', status: 'paid' }))
})
@@ -0,0 +1,67 @@
import { createHash } from 'node:crypto'
import type { QueryResult, QueryResultRow } from 'pg'
import { safeJsonStringify } from '../utils/json.js'
type QueryExecutor = <T extends QueryResultRow = QueryResultRow>(
text: string,
params?: unknown[],
) => Promise<QueryResult<T>>
export type OrderSourceSnapshotInput = {
orderId: number
provider: string
platform: string
shopId: string
platformOrderId: string
sourceLabel: string
payload: unknown
recordedAt: string
}
export function orderSourceSnapshotHash(payload: unknown): string {
return createHash('sha256')
.update(safeJsonStringify(payload || {}))
.digest('hex')
}
/**
* 追加平台订单原始快照。
* 同一订单的相同内容只写一次,不同内容永远追加,不覆盖订单主表原始报文。
*/
export async function appendOrderSourceSnapshotWithExecutor(
executor: QueryExecutor,
input: OrderSourceSnapshotInput,
): Promise<boolean> {
const payloadJson = safeJsonStringify(input.payload || {})
const payloadHash = orderSourceSnapshotHash(input.payload)
const result = await executor<{ id: number }>(
`
INSERT INTO order_source_snapshots (
order_id,
provider,
platform,
shop_id,
platform_order_id,
source_label,
payload_json,
payload_sha256,
recorded_at
) VALUES ($1, $2, $3, $4, $5, $6, $7::jsonb, $8, $9)
ON CONFLICT (order_id, payload_sha256) DO NOTHING
RETURNING id
`,
[
input.orderId,
input.provider,
input.platform,
input.shopId,
input.platformOrderId,
input.sourceLabel,
payloadJson,
payloadHash,
input.recordedAt,
],
)
return result.rows.length > 0
}
@@ -223,6 +223,22 @@ test('内部现货不再因上一手 last_assigned 进入打手可见集', async
assert.equal(calls.length, 2, '列表和概览必须共用同一条历史可见条件') assert.equal(calls.length, 2, '列表和概览必须共用同一条历史可见条件')
}) })
test('后台接单人员筛选把寄售供货人的绑定工单纳入可见集合', () => {
const query = buildWorkOrderWhere({
workerSharingId: 1494,
consignmentSupplierWorkerId: 1494,
})
assert.match(query.joins, /FROM consignment_inventories supplier_ci/)
assert.match(query.joins, /supplier_ci\.worker_id = \$2::bigint/)
assert.match(query.joins, /supplier_ci\.bound_work_order_id IS NOT NULL/)
assert.deepEqual(query.params, [1494, 1494])
// 打手端「我的订单」不传扩展口径,可见范围必须保持原样
const plain = buildWorkOrderWhere({ workerSharingId: 1494 })
assert.doesNotMatch(plain.joins, /supplier_ci/)
assert.deepEqual(plain.params, [1494])
})
test('「仅问题单」检索走短路分支,从 pending 表反查而非枚举可见集', () => { test('「仅问题单」检索走短路分支,从 pending 表反查而非枚举可见集', () => {
const query = buildWorkOrderWhere({ workerSharingId: 7, statuses: ['problem'] }) const query = buildWorkOrderWhere({ workerSharingId: 7, statuses: ['problem'] })
@@ -64,6 +64,52 @@ test('寄卖仓库搜索必须覆盖供货打手姓名、打手电话与页面
) )
}) })
test('寄卖提报单搜索必须能搜打手,列表与计数 SQL 都带实时打手关联', async () => {
const source = await readFile(new URL('./consignment-repo.ts', import.meta.url), 'utf8')
const start = source.indexOf('export async function listConsignmentSubmissions')
assert.ok(start > 0, '未找到寄卖提报单列表查询')
const body = source.slice(start, source.indexOf('\n}\n', start))
for (const field of [
'cs.worker_name',
'cs.worker_phone',
'supplier_worker.display_name',
'supplier_worker.username',
'supplier_worker.phone',
]) {
assert.ok(body.includes(field), `提报单搜索缺少字段:${field}`)
}
assert.ok(body.includes('AS live_worker_name'), '列表必须带出实时打手昵称')
assert.ok(body.includes('AS live_worker_phone'), '列表必须带出实时打手电话')
const countSqlLine = body
.split('\n')
.find((line) =>
line.includes('SELECT COUNT(*)::integer AS total FROM consignment_submissions cs'),
)
assert.ok(countSqlLine, '未找到提报单计数 SQL')
assert.ok(
countSqlLine!.includes('LEFT JOIN worker_users supplier_worker'),
'计数 SQL 缺少打手 join,按姓名/电话搜索会报错',
)
})
test('寄售库存详情/批量/匹配候选查询都带出实时打手昵称', async () => {
const source = await readFile(new URL('./consignment-repo.ts', import.meta.url), 'utf8')
for (const fn of [
'getConsignmentInventoryById',
'listConsignmentInventoriesByIds',
'getConsignmentInventoryBySubmissionId',
'listActiveConsignmentInventoriesForMatching',
]) {
const start = source.indexOf(`export async function ${fn}`)
assert.ok(start > 0, `未找到查询:${fn}`)
const body = source.slice(start, source.indexOf('\n}\n', start))
assert.ok(body.includes('LEFT JOIN worker_users supplier_worker'), `${fn} 缺少实时打手 join`)
assert.ok(body.includes('AS live_worker_name'), `${fn} 缺少实时打手昵称字段`)
}
})
test('寄卖仓库按通过审核入库时间筛选,结束日期必须包含当天全天', async () => { test('寄卖仓库按通过审核入库时间筛选,结束日期必须包含当天全天', async () => {
const source = await readRepoSource() const source = await readRepoSource()
const inventoryListStart = source.indexOf('export async function listConsignmentInventories') const inventoryListStart = source.indexOf('export async function listConsignmentInventories')
@@ -54,6 +54,10 @@ export type ConsignmentInventoryRow = {
worker_id?: string | number | null worker_id?: string | number | null
worker_name?: string worker_name?: string
worker_phone?: string worker_phone?: string
/** 当前打手昵称(实时 join worker_users,优先于入库时的快照) */
live_worker_name?: string | null
/** 当前打手电话(实时 join worker_users,优先于入库时的快照) */
live_worker_phone?: string | null
assigned_worker_id?: string | number | null assigned_worker_id?: string | number | null
deposit_amount_fen?: number deposit_amount_fen?: number
/** 分类级寄卖配置与发布快照 */ /** 分类级寄卖配置与发布快照 */
@@ -104,6 +108,10 @@ export type ConsignmentSubmissionRow = {
worker_id?: string | number | null worker_id?: string | number | null
worker_name?: string worker_name?: string
worker_phone?: string worker_phone?: string
/** 当前打手昵称(实时 join worker_users,优先于提报时的快照) */
live_worker_name?: string | null
/** 当前打手电话(实时 join worker_users,优先于提报时的快照) */
live_worker_phone?: string | null
deposit_amount_fen?: number deposit_amount_fen?: number
/** 分类级寄卖配置与发布快照 */ /** 分类级寄卖配置与发布快照 */
profile_id?: string | number | null profile_id?: string | number | null
@@ -324,6 +332,8 @@ export async function listConsignmentInventories(filters: {
wc.name AS category_name, wc.name AS category_name,
cs.submission_no AS source_submission_no, cs.submission_no AS source_submission_no,
cs.screenshots_json AS submission_screenshots_json, cs.screenshots_json AS submission_screenshots_json,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
wo.acceptance_json AS gift_evidence_json, wo.acceptance_json AS gift_evidence_json,
wo.work_order_no AS gift_work_order_no, wo.work_order_no AS gift_work_order_no,
wo.status AS gift_work_order_status, wo.status AS gift_work_order_status,
@@ -487,6 +497,8 @@ export async function getConsignmentInventoryById(
wc.name AS category_name, wc.name AS category_name,
cs.submission_no AS source_submission_no, cs.submission_no AS source_submission_no,
cs.screenshots_json AS submission_screenshots_json, cs.screenshots_json AS submission_screenshots_json,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
wo.acceptance_json AS gift_evidence_json, wo.acceptance_json AS gift_evidence_json,
wo.work_order_no AS gift_work_order_no, wo.work_order_no AS gift_work_order_no,
wo.status AS gift_work_order_status, wo.status AS gift_work_order_status,
@@ -498,6 +510,7 @@ export async function getConsignmentInventoryById(
FROM consignment_inventories ci FROM consignment_inventories ci
LEFT JOIN work_categories wc ON wc.id = ci.category_id LEFT JOIN work_categories wc ON wc.id = ci.category_id
LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = ci.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id
LEFT JOIN work_orders wo ON wo.id = ci.bound_work_order_id LEFT JOIN work_orders wo ON wo.id = ci.bound_work_order_id
WHERE ci.id = $1 WHERE ci.id = $1
@@ -519,10 +532,13 @@ export async function listConsignmentInventoriesByIds(
wc.name AS category_name, wc.name AS category_name,
cs.submission_no AS source_submission_no, cs.submission_no AS source_submission_no,
cs.screenshots_json AS submission_screenshots_json, cs.screenshots_json AS submission_screenshots_json,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
ccs.name AS contact_supporter_name ccs.name AS contact_supporter_name
FROM consignment_inventories ci FROM consignment_inventories ci
LEFT JOIN work_categories wc ON wc.id = ci.category_id LEFT JOIN work_categories wc ON wc.id = ci.category_id
LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = ci.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id
WHERE ci.id = ANY($1::bigint[]) WHERE ci.id = ANY($1::bigint[])
` `
@@ -1150,9 +1166,12 @@ export async function listActiveConsignmentInventoriesForMatching(options?: {
SELECT SELECT
ci.*, ci.*,
cs.submission_no AS source_submission_no, cs.submission_no AS source_submission_no,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
ccs.name AS contact_supporter_name ccs.name AS contact_supporter_name
FROM consignment_inventories ci FROM consignment_inventories ci
LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = ci.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id
WHERE ${conditions.join(' AND ')} WHERE ${conditions.join(' AND ')}
ORDER BY ci.id ASC ORDER BY ci.id ASC
@@ -1184,12 +1203,22 @@ export async function listConsignmentSubmissions(filters: {
if (filters.keyword) { if (filters.keyword) {
values.push(`%${filters.keyword.trim()}%`) values.push(`%${filters.keyword.trim()}%`)
conditions.push( conditions.push(
`(cs.item_name ILIKE $${values.length} OR cs.submission_no ILIKE $${values.length} OR cs.contact_info ILIKE $${values.length} OR cs.seller_game_id ILIKE $${values.length} OR cs.seller_phone ILIKE $${values.length} OR ccs.name ILIKE $${values.length})`, `(cs.item_name ILIKE $${values.length}
OR cs.submission_no ILIKE $${values.length}
OR cs.contact_info ILIKE $${values.length}
OR cs.seller_game_id ILIKE $${values.length}
OR cs.seller_phone ILIKE $${values.length}
OR cs.worker_name ILIKE $${values.length}
OR cs.worker_phone ILIKE $${values.length}
OR supplier_worker.display_name ILIKE $${values.length}
OR supplier_worker.username ILIKE $${values.length}
OR supplier_worker.phone ILIKE $${values.length}
OR ccs.name ILIKE $${values.length})`,
) )
} }
const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : '' const whereClause = conditions.length > 0 ? `WHERE ${conditions.join(' AND ')}` : ''
const countSql = `SELECT COUNT(*)::integer AS total FROM consignment_submissions cs LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id ${whereClause}` const countSql = `SELECT COUNT(*)::integer AS total FROM consignment_submissions cs LEFT JOIN worker_users supplier_worker ON supplier_worker.id = cs.worker_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id ${whereClause}`
const countRes = await query<{ total: number }>(countSql, values) const countRes = await query<{ total: number }>(countSql, values)
const total = countRes.rows[0]?.total || 0 const total = countRes.rows[0]?.total || 0
@@ -1206,9 +1235,12 @@ export async function listConsignmentSubmissions(filters: {
wo.acceptance_json AS gift_evidence_json, wo.acceptance_json AS gift_evidence_json,
wo.work_order_no AS gift_work_order_no, wo.work_order_no AS gift_work_order_no,
wo.status AS gift_work_order_status, wo.status AS gift_work_order_status,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
ccs.name AS contact_supporter_name ccs.name AS contact_supporter_name
FROM consignment_submissions cs FROM consignment_submissions cs
LEFT JOIN work_categories wc ON wc.id = cs.category_id LEFT JOIN work_categories wc ON wc.id = cs.category_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = cs.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id
LEFT JOIN consignment_inventories rinv ON (rinv.id = cs.result_inventory_id OR rinv.source_submission_id = cs.id) LEFT JOIN consignment_inventories rinv ON (rinv.id = cs.result_inventory_id OR rinv.source_submission_id = cs.id)
LEFT JOIN work_orders wo ON wo.id = rinv.bound_work_order_id LEFT JOIN work_orders wo ON wo.id = rinv.bound_work_order_id
@@ -1473,9 +1505,12 @@ export async function listConsignmentSubmissionsByIds(
wo.acceptance_json AS gift_evidence_json, wo.acceptance_json AS gift_evidence_json,
wo.work_order_no AS gift_work_order_no, wo.work_order_no AS gift_work_order_no,
wo.status AS gift_work_order_status, wo.status AS gift_work_order_status,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
ccs.name AS contact_supporter_name ccs.name AS contact_supporter_name
FROM consignment_submissions cs FROM consignment_submissions cs
LEFT JOIN work_categories wc ON wc.id = cs.category_id LEFT JOIN work_categories wc ON wc.id = cs.category_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = cs.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id
LEFT JOIN consignment_inventories rinv LEFT JOIN consignment_inventories rinv
ON (rinv.id = cs.result_inventory_id OR rinv.source_submission_id = cs.id) ON (rinv.id = cs.result_inventory_id OR rinv.source_submission_id = cs.id)
@@ -1502,10 +1537,13 @@ export async function getConsignmentInventoryBySubmissionId(
ci.*, ci.*,
wc.name AS category_name, wc.name AS category_name,
cs.submission_no AS source_submission_no, cs.submission_no AS source_submission_no,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
ccs.name AS contact_supporter_name ccs.name AS contact_supporter_name
FROM consignment_inventories ci FROM consignment_inventories ci
LEFT JOIN work_categories wc ON wc.id = ci.category_id LEFT JOIN work_categories wc ON wc.id = ci.category_id
LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id LEFT JOIN consignment_submissions cs ON cs.id = ci.source_submission_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = ci.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = ci.contact_supporter_id
WHERE ci.source_submission_id = $1 WHERE ci.source_submission_id = $1
ORDER BY ci.id DESC ORDER BY ci.id DESC
@@ -1526,9 +1564,12 @@ export async function getConsignmentSubmissionById(
wo.acceptance_json AS gift_evidence_json, wo.acceptance_json AS gift_evidence_json,
wo.work_order_no AS gift_work_order_no, wo.work_order_no AS gift_work_order_no,
wo.status AS gift_work_order_status, wo.status AS gift_work_order_status,
COALESCE(NULLIF(supplier_worker.display_name, ''), supplier_worker.username) AS live_worker_name,
supplier_worker.phone AS live_worker_phone,
ccs.name AS contact_supporter_name ccs.name AS contact_supporter_name
FROM consignment_submissions cs FROM consignment_submissions cs
LEFT JOIN work_categories wc ON wc.id = cs.category_id LEFT JOIN work_categories wc ON wc.id = cs.category_id
LEFT JOIN worker_users supplier_worker ON supplier_worker.id = cs.worker_id
LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id LEFT JOIN consignment_contact_supporters ccs ON ccs.id = cs.contact_supporter_id
LEFT JOIN consignment_inventories rinv ON (rinv.id = cs.result_inventory_id OR rinv.source_submission_id = cs.id) LEFT JOIN consignment_inventories rinv ON (rinv.id = cs.result_inventory_id OR rinv.source_submission_id = cs.id)
LEFT JOIN work_orders wo ON wo.id = rinv.bound_work_order_id LEFT JOIN work_orders wo ON wo.id = rinv.bound_work_order_id
@@ -529,6 +529,14 @@ export type ListInput = {
/** 后台打手列表按提现黑名单筛选。 */ /** 后台打手列表按提现黑名单筛选。 */
isWithdrawalBlacklisted?: boolean | null | undefined isWithdrawalBlacklisted?: boolean | null | undefined
workerSharingId?: number workerSharingId?: number
/**
* 后台「接单人员」筛选的扩展口径:同时把该打手作为寄售供货人的工单纳入可见集合。
*
* 内部寄售直接履约不写 assigned_worker_id,供货关系只存在
* consignment_inventories.worker_id;仅靠 workerSharingId 会漏掉这些单。
* 该字段只由后台筛选使用,打手端「我的订单」不传,避免改变打手可见范围。
*/
consignmentSupplierWorkerId?: number
/** 大厅候选集合的总展示上限;筛选与分页在候选集合内执行。 */ /** 大厅候选集合的总展示上限;筛选与分页在候选集合内执行。 */
hallCandidateLimit?: number hallCandidateLimit?: number
hallCategoryLimits?: Record<string, number> hallCategoryLimits?: Record<string, number>
@@ -391,6 +391,7 @@ export async function listWorkOrders({
workerLevelIds = [], workerLevelIds = [],
categoryId = 0, categoryId = 0,
workerSharingId = 0, workerSharingId = 0,
consignmentSupplierWorkerId = 0,
hallCandidateLimit = 0, hallCandidateLimit = 0,
hallCategoryLimits = {}, hallCategoryLimits = {},
visibleAfterIso = '', visibleAfterIso = '',
@@ -416,6 +417,7 @@ export async function listWorkOrders({
...(resolvedWorkerIds ? { workerIds: resolvedWorkerIds } : {}), ...(resolvedWorkerIds ? { workerIds: resolvedWorkerIds } : {}),
categoryId, categoryId,
workerSharingId, workerSharingId,
consignmentSupplierWorkerId,
hallCandidateLimit, hallCandidateLimit,
hallCategoryLimits, hallCategoryLimits,
visibleAfterIso, visibleAfterIso,
@@ -1212,6 +1214,7 @@ export function buildWorkOrderWhere({
workerIds, workerIds,
categoryId = 0, categoryId = 0,
workerSharingId = 0, workerSharingId = 0,
consignmentSupplierWorkerId = 0,
hallCandidateLimit = 0, hallCandidateLimit = 0,
hallCategoryLimits = {}, hallCategoryLimits = {},
visibleAfterIso = '', visibleAfterIso = '',
@@ -1241,6 +1244,7 @@ export function buildWorkOrderWhere({
) )
} }
let workerSharePlaceholder = '' let workerSharePlaceholder = ''
let consignmentSupplierUnion = ''
// 旧的 excludeFilledSharing 写法:LEFT JOIN 一份「整张 work_order_shares 的聚合」 // 旧的 excludeFilledSharing 写法:LEFT JOIN 一份「整张 work_order_shares 的聚合」
// (GROUP BY work_order_id)再在 WHERE 里比较。列表/计数每次调用都要重算这份全局聚合, // (GROUP BY work_order_id)再在 WHERE 里比较。列表/计数每次调用都要重算这份全局聚合,
@@ -1252,6 +1256,22 @@ export function buildWorkOrderWhere({
if (workerSharingId) { if (workerSharingId) {
params.push(workerSharingId) params.push(workerSharingId)
workerSharePlaceholder = `$${params.length}` workerSharePlaceholder = `$${params.length}`
const normalizedSupplierWorkerId = Math.max(
0,
Math.floor(Number(consignmentSupplierWorkerId) || 0),
)
if (normalizedSupplierWorkerId > 0) {
// 内部寄售直接履约不写 assigned_worker_id:供货关系只存在
// consignment_inventories.worker_id,这里按库存回查绑定工单,补进可见集合。
params.push(normalizedSupplierWorkerId)
const supplierPlaceholder = `$${params.length}`
consignmentSupplierUnion = `
UNION
SELECT supplier_ci.bound_work_order_id AS work_order_id
FROM consignment_inventories supplier_ci
WHERE supplier_ci.worker_id = ${supplierPlaceholder}::bigint
AND supplier_ci.bound_work_order_id IS NOT NULL`
}
joins.push(` joins.push(`
INNER JOIN ( INNER JOIN (
SELECT assigned.id AS work_order_id SELECT assigned.id AS work_order_id
@@ -1270,7 +1290,7 @@ export function buildWorkOrderWhere({
SELECT timeout_ev.work_order_id SELECT timeout_ev.work_order_id
FROM work_order_events timeout_ev FROM work_order_events timeout_ev
WHERE timeout_ev.event_type LIKE 'timeout_%' WHERE timeout_ev.event_type LIKE 'timeout_%'
AND timeout_ev.payload_json->>'workerId' = ${workerSharePlaceholder}::text AND timeout_ev.payload_json->>'workerId' = ${workerSharePlaceholder}::text${consignmentSupplierUnion}
) worker_visible ON worker_visible.work_order_id = wo.id ) worker_visible ON worker_visible.work_order_id = wo.id
LEFT JOIN LATERAL ( LEFT JOIN LATERAL (
SELECT wos.status, wos.created_at SELECT wos.status, wos.created_at
@@ -7,6 +7,7 @@ import {
applyOrderItemRevenueAllocationWithExecutor, applyOrderItemRevenueAllocationWithExecutor,
replaceOrderItemsWithExecutor, replaceOrderItemsWithExecutor,
} from '../../repositories/order-item-repo.js' } from '../../repositories/order-item-repo.js'
import { appendOrderSourceSnapshotWithExecutor } from '../../repositories/order-source-snapshot-repo.js'
import { withTransaction } from '../../db/client.js' import { withTransaction } from '../../db/client.js'
import { syncDeliveryTasksForOrder } from './delivery-task-service.js' import { syncDeliveryTasksForOrder } from './delivery-task-service.js'
import { resolveOrderItemForFulfillment } from '../fulfillment/product-resolution-service.js' import { resolveOrderItemForFulfillment } from '../fulfillment/product-resolution-service.js'
@@ -189,6 +190,34 @@ export async function upsertOrderFromSource(
}) })
} }
await appendOrderSourceSnapshotWithExecutor(executor, {
orderId: Number(order.id),
provider: event.provider,
platform: event.platform,
shopId: String(order.shop_id || event.shopId || '').trim(),
platformOrderId: event.platformOrderId,
sourceLabel,
payload: {
sourceLabel,
provider: event.provider,
platform: event.platform,
shopId: event.shopId,
shopName: event.shopName,
platformOrderId: event.platformOrderId,
orderStatus: event.orderStatus,
payStatus: event.payStatus,
buyerId: event.buyerId,
buyerName: event.buyerName,
receiverContact: event.receiverContact,
totalAmount: event.totalAmount,
currency: event.currency,
paidAt: event.paidAt || null,
rawPayload: event.rawPayload,
items: event.items,
},
recordedAt: now,
})
const orderItems = await replaceOrderItemsWithExecutor( const orderItems = await replaceOrderItemsWithExecutor(
executor, executor,
order.id, order.id,
@@ -25,9 +25,8 @@ import type { OrderItemRow, OrderRow } from '../../../types/repository/rows.js'
export const OPEN_91_PENDING_CONFIG_STATUS = 'pending_config' export const OPEN_91_PENDING_CONFIG_STATUS = 'pending_config'
export const OPEN_91_MANUAL_FAILED_STATUS = 'manual_failed' export const OPEN_91_MANUAL_FAILED_STATUS = 'manual_failed'
export async function markOpen91OrderPendingConfig(order: OrderRow, reason = '') { export async function markOpen91OrderPendingConfig(order: OrderRow, _reason = '') {
const now = nowIso() const now = nowIso()
const rawPayload = parseJsonObject(order.raw_payload_json)
return updateOrder(order.id, { return updateOrder(order.id, {
provider: order.provider, provider: order.provider,
platform: order.platform, platform: order.platform,
@@ -41,11 +40,6 @@ export async function markOpen91OrderPendingConfig(order: OrderRow, reason = '')
receiverContact: order.receiver_contact, receiverContact: order.receiver_contact,
totalAmount: Number(order.total_amount) || 0, totalAmount: Number(order.total_amount) || 0,
currency: order.currency, currency: order.currency,
rawPayloadJson: JSON.stringify({
...rawPayload,
pendingReason: String(reason || '').trim() || 'fulfillment_not_ready',
pendingAt: now,
}),
paidAt: order.paid_at, paidAt: order.paid_at,
updatedAt: now, updatedAt: now,
}) })
@@ -154,11 +148,6 @@ export async function upsertOpen91PendingOrder(payload: JsonObject = {}, config:
platformOrderId: event.platformOrderId, platformOrderId: event.platformOrderId,
}) })
const rawPayloadJson = JSON.stringify({
...event.rawPayload,
pendingReason: 'unconfigured_items',
pendingAt: now,
})
const orderPayload = { const orderPayload = {
provider: event.provider, provider: event.provider,
platform: event.platform, platform: event.platform,
@@ -172,7 +161,6 @@ export async function upsertOpen91PendingOrder(payload: JsonObject = {}, config:
receiverContact: '', receiverContact: '',
totalAmount: event.totalAmount, totalAmount: event.totalAmount,
currency: 'CNY', currency: 'CNY',
rawPayloadJson,
paidAt: existing?.paid_at || now, paidAt: existing?.paid_at || now,
} }
@@ -245,10 +233,9 @@ export async function retryOpen91Order(orderId: string | number) {
} }
} }
export async function failOpen91Order(orderId: string | number, reason = '') { export async function failOpen91Order(orderId: string | number, _reason = '') {
const order = await getRequiredOpen91Order(orderId) const order = await getRequiredOpen91Order(orderId)
const now = nowIso() const now = nowIso()
const rawPayload = parseJsonObject(order.raw_payload_json)
const updated = await updateOrder(order.id, { const updated = await updateOrder(order.id, {
provider: order.provider, provider: order.provider,
platform: order.platform, platform: order.platform,
@@ -262,11 +249,6 @@ export async function failOpen91Order(orderId: string | number, reason = '') {
receiverContact: order.receiver_contact, receiverContact: order.receiver_contact,
totalAmount: order.total_amount, totalAmount: order.total_amount,
currency: order.currency, currency: order.currency,
rawPayloadJson: JSON.stringify({
...rawPayload,
manualFailedAt: now,
manualFailedReason: String(reason || '').trim() || '商家手动标记无法履约',
}),
paidAt: order.paid_at, paidAt: order.paid_at,
updatedAt: now, updatedAt: now,
}) })
@@ -3,12 +3,14 @@ import type { Response } from 'express'
import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js' import { WORK_ORDER_STATUS } from '../../domain/work-order-status.js'
import { import {
type ConsignmentInventoryRow,
getLatestMaterialEventAtByOrderIds, getLatestMaterialEventAtByOrderIds,
getPendingCancelRequestOrderIds, getPendingCancelRequestOrderIds,
getPendingFeedbackOrderIds, getPendingFeedbackOrderIds,
getWorkerFinanceRequestById, getWorkerFinanceRequestById,
getWorkerUserById, getWorkerUserById,
getWorkOrderById, getWorkOrderById,
listConsignmentInventoriesByIds,
listWorkerTimeoutEventsByOrderIds, listWorkerTimeoutEventsByOrderIds,
listWorkerWorkOrderNotes, listWorkerWorkOrderNotes,
listWorkerWorkOrderViews, listWorkerWorkOrderViews,
@@ -23,6 +25,7 @@ import {
type RealtimeEventRow, type RealtimeEventRow,
} from '../../repositories/realtime-event-repo.js' } from '../../repositories/realtime-event-repo.js'
import { getAdminDashboardSummary } from '../admin/admin-dashboard-service.js' import { getAdminDashboardSummary } from '../admin/admin-dashboard-service.js'
import { safeParseJson } from '../admin/admin-query-utils.js'
import { mapAdminOrderListItem } from '../admin/admin-order-read-helpers.js' import { mapAdminOrderListItem } from '../admin/admin-order-read-helpers.js'
import { mapAdminTaskListItem } from '../admin/admin-task-read-helpers.js' import { mapAdminTaskListItem } from '../admin/admin-task-read-helpers.js'
import { import {
@@ -298,8 +301,27 @@ async function publishWorkOrderSnapshots(input: {
return return
} }
const pendingUnfreezeByOrderId = await sumPendingUnfreezeByOrderIds([workOrderId]) const material = safeParseJson(workOrder.material_json)
const shares = await listWorkOrderShares(workOrderId) const consignmentMaterial = material.consignment
const consignmentInventoryId =
consignmentMaterial &&
typeof consignmentMaterial === 'object' &&
!Array.isArray(consignmentMaterial)
? Math.max(
0,
Math.floor(Number((consignmentMaterial as Record<string, unknown>).inventoryId) || 0),
)
: 0
const [pendingUnfreezeByOrderId, shares, consignmentInventories] = await Promise.all([
sumPendingUnfreezeByOrderIds([workOrderId]),
listWorkOrderShares(workOrderId),
consignmentInventoryId > 0
? listConsignmentInventoriesByIds([consignmentInventoryId])
: Promise.resolve([] as ConsignmentInventoryRow[]),
])
// 寄售工单的快照也要带库存实时供货人:前端会整行替换,
// 若只给 material 里的旧快照,改名后的列表会被实时事件刷回旧名。
const consignmentInventory = consignmentInventories[0] || null
// 事件驱动缓存失效:这里已经拿到的工单与拼单快照恰好给出完整的受影响打手集合 // 事件驱动缓存失效:这里已经拿到的工单与拼单快照恰好给出完整的受影响打手集合
// (当前指派 + 历史保留 + 全部拼单人 + 调用方显式指定的)。 // (当前指派 + 历史保留 + 全部拼单人 + 调用方显式指定的)。
@@ -317,7 +339,7 @@ async function publishWorkOrderSnapshots(input: {
.reduce((sum, share) => sum + Number(share.quantity || 0), 0) .reduce((sum, share) => sum + Number(share.quantity || 0), 0)
const pendingSubmissionCount = shares.filter((share) => share.status === 'joined').length const pendingSubmissionCount = shares.filter((share) => share.status === 'joined').length
const adminSnapshot = { const adminSnapshot = {
...mapWorkOrderAdmin(workOrder, shares), ...mapWorkOrderAdmin(workOrder, shares, consignmentInventory),
pendingUnfreezeAmount: pendingUnfreezeByOrderId.get(workOrderId) || 0, pendingUnfreezeAmount: pendingUnfreezeByOrderId.get(workOrderId) || 0,
sharingProgress: { sharingProgress: {
joinedQuantity, joinedQuantity,
@@ -6,21 +6,25 @@ async function readJobSource() {
return readFile(new URL('./data-retention-job.ts', import.meta.url), 'utf8') return readFile(new URL('./data-retention-job.ts', import.meta.url), 'utf8')
} }
test('保留清理必须覆盖 work_order_events,否则这张只增不减的表会一直拖慢历史单检索', async () => { test('数据维护不得覆盖原始报文或删除永久业务事实', async () => {
const src = await readJobSource() const src = await readJobSource()
assert.ok( assert.equal(src.includes('raw_payload_json ='), false)
src.includes("deleteOldRowsInBatches('work_order_events'"), assert.equal(src.includes('DELETE FROM'), false)
'work_order_events 未纳入保留清理(生产上 71 万行 / 269MB 且只增不减)', for (const table of [
) 'orders',
assert.ok( 'kuaishou_industry_vouchers',
src.includes("'task_events' | 'webhook_events' | 'admin_audit_logs' | 'work_order_events'"), 'task_events',
'删除助手需要显式允许 work_order_events(表名是白名单联合类型,避免拼接注入)', 'webhook_events',
) 'admin_audit_logs',
assert.ok(src.includes('retention.workOrderEventDays'), '保留天数必须走配置项,便于按环境调整') 'work_order_events',
assert.ok( ]) {
src.includes('workOrderEventRows: number') && src.includes('workOrderEventRows,'), assert.equal(
'清理结果要带上工单事件的删除行数,方便定时任务面板核对', new RegExp(`(?:UPDATE|DELETE)\\s+${table}\\b`, 'i').test(src),
) false,
`${table} 属于永久业务事实,不能在维护任务中清理或覆盖`,
)
}
assert.ok(src.includes('永久业务事实未清理'))
}) })
test('成本异常只归档已处理的行,open 队列不能被藏起来', async () => { test('成本异常只归档已处理的行,open 队列不能被藏起来', async () => {
@@ -64,29 +68,28 @@ test('保留任务要回填漏写的付款派生列,并在有命中时打日
) )
}) })
test('保留清理默认窗口:敏感报文/事件 180 天,审计 365 天,工单事件 180 天', async () => { test('数据维护只保留可派生数据的成本异常归档窗口', async () => {
const defaults = await readFile(new URL('../../config/defaults.ts', import.meta.url), 'utf8') const defaults = await readFile(new URL('../../config/defaults.ts', import.meta.url), 'utf8')
assert.ok(defaults.includes('costExceptionDays: 60'))
for (const line of [ for (const line of [
'rawPayloadDays: 180', 'rawPayloadDays:',
'taskEventDays: 180', 'taskEventDays:',
'webhookEventDays: 180', 'webhookEventDays:',
'auditLogDays: 365', 'auditLogDays:',
'workOrderEventDays: 180', 'workOrderEventDays:',
'costExceptionDays: 60',
]) { ]) {
assert.ok(defaults.includes(line), `默认保留策略缺少 ${line}`) assert.equal(defaults.includes(line), false, `永久事实不能存在清理窗口 ${line}`)
} }
const overrides = await readFile( const overrides = await readFile(
new URL('../../config/env-overrides.ts', import.meta.url), new URL('../../config/env-overrides.ts', import.meta.url),
'utf8', 'utf8',
) )
assert.ok( assert.equal(overrides.includes('RAW_PAYLOAD_RETENTION_DAYS'), false)
overrides.includes( assert.equal(overrides.includes('TASK_EVENT_RETENTION_DAYS'), false)
"integerEnv('WORK_ORDER_EVENT_RETENTION_DAYS', ['retention', 'workOrderEventDays'])", assert.equal(overrides.includes('WEBHOOK_EVENT_RETENTION_DAYS'), false)
), assert.equal(overrides.includes('AUDIT_LOG_RETENTION_DAYS'), false)
'WORK_ORDER_EVENT_RETENTION_DAYS 需要环境变量出口,否则生产无法调整', assert.equal(overrides.includes('WORK_ORDER_EVENT_RETENTION_DAYS'), false)
)
assert.ok( assert.ok(
overrides.includes( overrides.includes(
"integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays'])", "integerEnv('COST_EXCEPTION_RETENTION_DAYS', ['retention', 'costExceptionDays'])",
@@ -8,49 +8,29 @@ import { pruneWorkProductMatchLogs } from '../../repositories/worker-platform/pr
export type DataRetentionResult = { export type DataRetentionResult = {
status: 'ok' status: 'ok'
message: string message: string
rawPayloadRows: number
taskEventRows: number
webhookEventRows: number
auditLogRows: number
workOrderEventRows: number
matchLogRows: number
matchLogDeletedRows: number matchLogDeletedRows: number
sourceOrderAmountBackfillRows: number sourceOrderAmountBackfillRows: number
costExceptionArchivedRows: number costExceptionArchivedRows: number
} }
/** 清理敏感原始报文并删除已过保留期的事件/审计记录。 */ /**
* 维护可重建的派生数据。
*
* 原始订单、平台报文、工单/任务事件、审计日志和资金记录都是永久事实,
* 不在这里按时间清理或覆盖。
*/
export async function runDataRetentionJob(): Promise<DataRetentionResult> { export async function runDataRetentionJob(): Promise<DataRetentionResult> {
const retention = runtimeConfig.retention || {} const retention = runtimeConfig.retention || {}
const rawPayloadDays = positiveDays(retention.rawPayloadDays, 180)
const taskEventDays = positiveDays(retention.taskEventDays, 180)
const webhookEventDays = positiveDays(retention.webhookEventDays, 180)
const auditLogDays = positiveDays(retention.auditLogDays, 365)
// 工单事件表以前完全没纳入保留:717k 行 / 269MB 且只增不减,会一直拖慢历史单检索。
const workOrderEventDays = positiveDays(retention.workOrderEventDays, 180)
// 成本异常只归档「已处理」的行;open 行是待办队列,不能因为超期就藏起来。 // 成本异常只归档「已处理」的行;open 行是待办队列,不能因为超期就藏起来。
const costExceptionDays = positiveDays(retention.costExceptionDays, 60) const costExceptionDays = positiveDays(retention.costExceptionDays, 60)
// 1. 按配置清理超出上限或过期的打手商品匹配日志 // 商品匹配日志是诊断数据,不是订单或履约事实。
const matchConfig = getWorkerProductMatchConfig() const matchConfig = getWorkerProductMatchConfig()
const matchLogPruneResult = await pruneWorkProductMatchLogs({ const matchLogPruneResult = await pruneWorkProductMatchLogs({
maxRows: matchConfig.maxMatchLogs, maxRows: matchConfig.maxMatchLogs,
retentionDays: matchConfig.matchLogRetentionDays, retentionDays: matchConfig.matchLogRetentionDays,
}) })
// 2. 清理其他报文与日志
const rawPayloadRows =
(await scrubJsonColumnInBatches('orders', 'updated_at', rawPayloadDays)) +
(await scrubJsonColumnInBatches('kuaishou_industry_vouchers', 'updated_at', rawPayloadDays))
const matchLogRows = await scrubJsonColumnInBatches(
'work_product_match_logs',
'created_at',
rawPayloadDays,
)
const taskEventRows = await deleteOldRowsInBatches('task_events', taskEventDays)
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。若将来有第三个写入点忘了维护派生列, // 统计热路径已不再读 material_json。若将来有第三个写入点忘了维护派生列,
// 金额会静默回退 orders.total_amount。这里把漏写补上并打日志,让口径变化可见。 // 金额会静默回退 orders.total_amount。这里把漏写补上并打日志,让口径变化可见。
const costExceptionArchive = await archiveResolvedCostExceptions({ const costExceptionArchive = await archiveResolvedCostExceptions({
@@ -66,14 +46,8 @@ export async function runDataRetentionJob(): Promise<DataRetentionResult> {
return { return {
status: 'ok', status: 'ok',
message: '数据保留清理完成', message: '派生数据维护完成;永久业务事实未清理',
rawPayloadRows,
matchLogRows,
matchLogDeletedRows: matchLogPruneResult.totalDeleted, matchLogDeletedRows: matchLogPruneResult.totalDeleted,
taskEventRows,
webhookEventRows,
auditLogRows,
workOrderEventRows,
sourceOrderAmountBackfillRows, sourceOrderAmountBackfillRows,
costExceptionArchivedRows, costExceptionArchivedRows,
} }
@@ -92,59 +66,6 @@ async function backfillMissingSourceOrderAmount() {
return result.rowCount || 0 return result.rowCount || 0
} }
async function scrubJsonColumnInBatches(
table: 'orders' | 'kuaishou_industry_vouchers' | 'work_product_match_logs',
dateColumn: 'created_at' | 'updated_at',
retentionDays: number,
) {
let total = 0
while (true) {
const result = await query(
`
WITH candidates AS (
SELECT ctid
FROM ${table}
WHERE ${dateColumn} < NOW() - make_interval(days => $1::int)
AND raw_payload_json <> '{}'::jsonb
LIMIT 1000
)
UPDATE ${table} AS target
SET raw_payload_json = '{}'::jsonb
FROM candidates
WHERE target.ctid = candidates.ctid
`,
[retentionDays],
)
total += result.rowCount || 0
if ((result.rowCount || 0) < 1000) return total
}
}
async function deleteOldRowsInBatches(
table: 'task_events' | 'webhook_events' | 'admin_audit_logs' | 'work_order_events',
retentionDays: number,
) {
let total = 0
while (true) {
const result = await query(
`
WITH candidates AS (
SELECT ctid
FROM ${table}
WHERE created_at < NOW() - make_interval(days => $1::int)
LIMIT 1000
)
DELETE FROM ${table} AS target
USING candidates
WHERE target.ctid = candidates.ctid
`,
[retentionDays],
)
total += result.rowCount || 0
if ((result.rowCount || 0) < 1000) return total
}
}
function positiveDays(value: unknown, fallback: number) { function positiveDays(value: unknown, fallback: number) {
const days = Number(value) const days = Number(value)
return Number.isInteger(days) && days > 0 ? days : fallback return Number.isInteger(days) && days > 0 ? days : fallback
@@ -169,12 +169,9 @@ async function runJob(job: JsonObject, { manual = false }: { manual?: boolean }
status, status,
...(job.type === 'data_retention' ...(job.type === 'data_retention'
? { ? {
rawPayloadRows: Number(summary.rawPayloadRows || 0), matchLogDeletedRows: Number(summary.matchLogDeletedRows || 0),
matchLogRows: Number(summary.matchLogRows || 0),
taskEventRows: Number(summary.taskEventRows || 0),
webhookEventRows: Number(summary.webhookEventRows || 0),
auditLogRows: Number(summary.auditLogRows || 0),
sourceOrderAmountBackfillRows: Number(summary.sourceOrderAmountBackfillRows || 0), sourceOrderAmountBackfillRows: Number(summary.sourceOrderAmountBackfillRows || 0),
costExceptionArchivedRows: Number(summary.costExceptionArchivedRows || 0),
} }
: {}), : {}),
}) })
@@ -59,6 +59,7 @@ function buildWorkOrderStatisticsCacheKey(filters: JsonObject): string {
filters.keyword || '', filters.keyword || '',
filters.shopNames || [], filters.shopNames || [],
filters.workerSharingId || 0, filters.workerSharingId || 0,
filters.consignmentSupplierWorkerId || 0,
filters.workerId || 0, filters.workerId || 0,
filters.workerLevelIds || [], filters.workerLevelIds || [],
filters.categoryId || 0, filters.categoryId || 0,
@@ -149,6 +150,9 @@ export function resolveAdminWorkOrderListFilters(query: JsonObject = {}): ListIn
keyword, keyword,
shopNames, shopNames,
workerSharingId: workerId, workerSharingId: workerId,
// 「接单人员」筛选同时覆盖寄售供货人:内部寄售直接履约不写 assigned_worker_id,
// 供货关系只存在 consignment_inventories.worker_id,必须补进可见集合。
...(workerId > 0 ? { consignmentSupplierWorkerId: workerId } : {}),
workerLevelIds, workerLevelIds,
vipEvidencePending, vipEvidencePending,
...(acceptanceMode ? { acceptanceMode } : {}), ...(acceptanceMode ? { acceptanceMode } : {}),
@@ -1,7 +1,7 @@
import assert from 'node:assert/strict' import assert from 'node:assert/strict'
import test from 'node:test' import test from 'node:test'
import { normalizeInventoryDate } from './consignment-admin-service.js' import { mapConsignmentInventory, normalizeInventoryDate } from './consignment-admin-service.js'
test('寄售仓库入库日期筛选只接受合法的 YYYY-MM-DD', () => { test('寄售仓库入库日期筛选只接受合法的 YYYY-MM-DD', () => {
assert.equal(normalizeInventoryDate('2026-11-05'), '2026-11-05') assert.equal(normalizeInventoryDate('2026-11-05'), '2026-11-05')
@@ -10,3 +10,61 @@ test('寄售仓库入库日期筛选只接受合法的 YYYY-MM-DD', () => {
assert.equal(normalizeInventoryDate('2026/11/05'), undefined) assert.equal(normalizeInventoryDate('2026/11/05'), undefined)
assert.equal(normalizeInventoryDate(''), undefined) assert.equal(normalizeInventoryDate(''), undefined)
}) })
test('寄售库存改名后优先显示实时打手昵称,打手被删才回退快照', () => {
const base = {
id: 1,
inventory_no: 'INV1',
game_name: '和平精英',
category_id: null,
item_name: '测试物品',
system: '',
server_zone: '',
series: '',
diamond_level: 1,
brilliance_count: 0,
cost_price_fen: 0,
buyer_name: '',
inbound_batch: '',
seller_contact: '',
seller_game_id: '',
seller_payment_info: '',
seller_phone: '',
credentials_json: {},
screenshots_json: [],
remark: '',
allow_matching: true,
status: 'in_stock',
allocation_status: '待安排',
bound_order_id: null,
bound_work_order_id: null,
bound_platform_order_id: '',
shop_name: '',
source_type: 'manual',
source_submission_id: null,
inbound_at: '2026-11-01T00:00:00.000Z',
consumed_at: null,
created_at: '2026-11-01T00:00:00.000Z',
updated_at: '2026-11-01T00:00:00.000Z',
worker_id: 7,
worker_name: '旧昵称',
worker_phone: '111',
}
const renamed = mapConsignmentInventory({
...base,
live_worker_name: '新昵称',
live_worker_phone: '222',
} as never)
assert.equal(renamed.workerName, '新昵称')
assert.equal(renamed.workerPhone, '222')
// 打手被删除时 join 不到实时行,仍要能显示登记时快照
const deleted = mapConsignmentInventory({
...base,
live_worker_name: null,
live_worker_phone: null,
} as never)
assert.equal(deleted.workerName, '旧昵称')
assert.equal(deleted.workerPhone, '111')
})
@@ -309,8 +309,9 @@ export function mapConsignmentInventory(row: ConsignmentInventoryRow) {
}) })
: null, : null,
workerId: row.worker_id ? Number(row.worker_id) : null, workerId: row.worker_id ? Number(row.worker_id) : null,
workerName: row.worker_name || '', // 改名后以当前昵称为准;快照只用于打手被删除等无法实时关联的兜底。
workerPhone: row.worker_phone || '', workerName: row.live_worker_name || row.worker_name || '',
workerPhone: row.live_worker_phone || row.worker_phone || '',
assignedWorkerId: row.assigned_worker_id ? Number(row.assigned_worker_id) : null, assignedWorkerId: row.assigned_worker_id ? Number(row.assigned_worker_id) : null,
depositAmountFen: Number(row.deposit_amount_fen || 0), depositAmountFen: Number(row.deposit_amount_fen || 0),
contactSupporterId: row.contact_supporter_id ? Number(row.contact_supporter_id) : null, contactSupporterId: row.contact_supporter_id ? Number(row.contact_supporter_id) : null,
@@ -371,8 +372,9 @@ export function mapConsignmentSubmission(row: ConsignmentSubmissionRow) {
), ),
askingPriceFen: Number(row.asking_price_fen || 0), askingPriceFen: Number(row.asking_price_fen || 0),
workerId: row.worker_id ? Number(row.worker_id) : null, workerId: row.worker_id ? Number(row.worker_id) : null,
workerName: row.worker_name || '', // 改名后以当前昵称为准;快照只用于打手被删除等无法实时关联的兜底。
workerPhone: row.worker_phone || '', workerName: row.live_worker_name || row.worker_name || '',
workerPhone: row.live_worker_phone || row.worker_phone || '',
depositAmountFen: Number(row.deposit_amount_fen || 0), depositAmountFen: Number(row.deposit_amount_fen || 0),
contactSupporterId: row.contact_supporter_id ? Number(row.contact_supporter_id) : null, contactSupporterId: row.contact_supporter_id ? Number(row.contact_supporter_id) : null,
contactSupporterName: row.contact_supporter_name || '', contactSupporterName: row.contact_supporter_name || '',
@@ -41,6 +41,7 @@ test('后台:显式单号检索时忽略状态等排他筛选,只按单号
assert.equal(filters.vipEvidencePending, false) assert.equal(filters.vipEvidencePending, false)
// 接单人员:目标单可能根本不在这名打手的可见集合里 // 接单人员:目标单可能根本不在这名打手的可见集合里
assert.equal(filters.workerSharingId, 0) assert.equal(filters.workerSharingId, 0)
assert.equal(filters.consignmentSupplierWorkerId, undefined)
// 分类 / 等级 / 验收模式 / 赠送阶段 / 创建时间 / 定位工单 // 分类 / 等级 / 验收模式 / 赠送阶段 / 创建时间 / 定位工单
assert.equal(filters.categoryId, 0) assert.equal(filters.categoryId, 0)
assert.deepEqual(filters.workerLevelIds, []) assert.deepEqual(filters.workerLevelIds, [])
@@ -77,6 +78,7 @@ test('后台:普通关键词(商品名/昵称)仍按原样下推全部筛
assert.deepEqual(filters.statuses, ['open', 'in_progress']) assert.deepEqual(filters.statuses, ['open', 'in_progress'])
assert.equal(filters.vipEvidencePending, true) assert.equal(filters.vipEvidencePending, true)
assert.equal(filters.workerSharingId, 7) assert.equal(filters.workerSharingId, 7)
assert.equal(filters.consignmentSupplierWorkerId, 7)
assert.equal(filters.categoryId, 12) assert.equal(filters.categoryId, 12)
assert.deepEqual(filters.workerLevelIds, [1, 2]) assert.deepEqual(filters.workerLevelIds, [1, 2])
assert.equal(filters.acceptanceMode, 'friend_gift') assert.equal(filters.acceptanceMode, 'friend_gift')
@@ -4,6 +4,7 @@ import test from 'node:test'
import type { WorkOrderShareRow } from '../../repositories/worker-platform/types.js' import type { WorkOrderShareRow } from '../../repositories/worker-platform/types.js'
import { import {
isWorkerOrderVipEvidencePending, isWorkerOrderVipEvidencePending,
mapWorkOrderMaterialWithConsignmentWorker,
pickWorkerRepresentativeShare, pickWorkerRepresentativeShare,
resolveWorkerOrderStatusTabs, resolveWorkerOrderStatusTabs,
stripWorkerPersonalShareFields, stripWorkerPersonalShareFields,
@@ -264,3 +265,36 @@ test('大厅广播快照必须剔除打手私有份额字段', () => {
assert.deepEqual(snapshot.sharingProgress, { joinedQuantity: 2, pendingSubmissionCount: 1 }) assert.deepEqual(snapshot.sharingProgress, { joinedQuantity: 2, pendingSubmissionCount: 1 })
assert.equal(snapshot.workOrderId, 1) assert.equal(snapshot.workOrderId, 1)
}) })
test('寄售供货人改名后,工单 material 展示使用库存实时昵称', () => {
const material = {
consignment: {
inventoryId: 9,
workerName: '旧昵称',
workerPhone: '111',
itemName: '测试物品',
},
}
const renamed = mapWorkOrderMaterialWithConsignmentWorker(material, {
live_worker_name: '新昵称',
live_worker_phone: '222',
} as never) as { consignment: { workerName: string; workerPhone: string; itemName: string } }
assert.equal(renamed.consignment.workerName, '新昵称')
assert.equal(renamed.consignment.workerPhone, '222')
// 其它寄售字段不能被覆盖
assert.equal(renamed.consignment.itemName, '测试物品')
const fallback = mapWorkOrderMaterialWithConsignmentWorker(material, null) as {
consignment: { workerName: string }
}
assert.equal(fallback.consignment.workerName, '旧昵称')
// 历史上没有登记供货人的内部现货单,不能因为库存关联到打手就改标成「供货打手」
const noSnapshot = mapWorkOrderMaterialWithConsignmentWorker(
{ consignment: { inventoryId: 9, workerName: '' } },
{ live_worker_name: '新昵称', live_worker_phone: '222' } as never,
) as { consignment: { workerName: string } }
assert.equal(noSnapshot.consignment.workerName, '')
})
@@ -68,6 +68,40 @@ export function mapWorkOrderMaterial(value: unknown) {
} }
} }
/**
* 工单 material.consignment 里保存的是入库/履约时的打手昵称快照。
* 打手改名后,后台列表仍要显示当前昵称;有实时库存行时用它覆盖展示字段,
* 历史 JSON 本身不改写(保留登记时痕迹)。
*/
export function mapWorkOrderMaterialWithConsignmentWorker(
value: unknown,
consignmentInventory?: ConsignmentInventoryRow | null,
) {
const material = mapWorkOrderMaterial(value) as JsonObject
const liveWorkerName = String(
consignmentInventory?.live_worker_name || consignmentInventory?.worker_name || '',
).trim()
if (!liveWorkerName) return material
const consignment = material.consignment
if (!consignment || typeof consignment !== 'object' || Array.isArray(consignment)) {
return material
}
const consignmentRecord = consignment as JsonObject
// 只刷新「原本就登记过供货人」的单据,避免把历史上没有供货人快照的
// 内部现货单改标成「供货打手」(那会改变历史单的展示语义)。
if (!String(consignmentRecord.workerName || '').trim()) return material
return {
...material,
consignment: {
...consignmentRecord,
workerName: liveWorkerName,
workerPhone: String(
consignmentInventory?.live_worker_phone || consignmentInventory?.worker_phone || '',
).trim(),
},
}
}
/** 读取好友赠送模式的打手本人资料;主页截图访问链接需要读取时重签。 */ /** 读取好友赠送模式的打手本人资料;主页截图访问链接需要读取时重签。 */
export function mapWorkOrderBoosterMaterial(value: unknown) { export function mapWorkOrderBoosterMaterial(value: unknown) {
const material = safeParseJson(value) const material = safeParseJson(value)
@@ -283,6 +317,12 @@ export function mapWorkOrderAdmin(
const giftPhase = String(workOrder.gift_phase || '').trim() const giftPhase = String(workOrder.gift_phase || '').trim()
const resolvedGiftPhase = resolveWorkOrderGiftPhase(workOrder) const resolvedGiftPhase = resolveWorkOrderGiftPhase(workOrder)
const internalConsignment = hasInternalConsignmentFulfillment(workOrder) const internalConsignment = hasInternalConsignmentFulfillment(workOrder)
const consignmentMaterial = safeParseJson(
safeParseJson(workOrder.material_json).consignment,
) as JsonObject
const consignmentWorkerIdRaw =
consignmentInventory?.worker_id || consignmentMaterial.workerId || 0
const consignmentWorkerId = Math.max(0, Math.floor(Number(consignmentWorkerIdRaw) || 0))
const hallAvailable = const hallAvailable =
workOrder.status === WORK_ORDER_STATUS.OPEN && workOrder.status === WORK_ORDER_STATUS.OPEN &&
(!workOrder.sharing_enabled || activeShareQuantity < sharingTotalQuantity) (!workOrder.sharing_enabled || activeShareQuantity < sharingTotalQuantity)
@@ -437,7 +477,10 @@ export function mapWorkOrderAdmin(
depositThresholdAmount: Number( depositThresholdAmount: Number(
workOrder.deposit_threshold_amount || DEFAULT_DEPOSIT_THRESHOLD_AMOUNT, workOrder.deposit_threshold_amount || DEFAULT_DEPOSIT_THRESHOLD_AMOUNT,
), ),
material: mapWorkOrderMaterial(workOrder.material_json), material: mapWorkOrderMaterialWithConsignmentWorker(
workOrder.material_json,
consignmentInventory,
),
requirement: { requirement: {
fields: resolveRequirementFields(workOrder), fields: resolveRequirementFields(workOrder),
}, },
@@ -462,6 +505,8 @@ export function mapWorkOrderAdmin(
displayName: workOrder.worker_display_name || '', displayName: workOrder.worker_display_name || '',
} }
: null, : null,
/** 寄售供货打手:后台「接单人员」筛选也认这个 id(内部寄售直接履约没有 assigned_worker_id)。 */
consignmentWorkerId: consignmentWorkerId > 0 ? consignmentWorkerId : null,
...(shares ? { sharingWorkers: mapWorkOrderSharingWorkers(shares) } : {}), ...(shares ? { sharingWorkers: mapWorkOrderSharingWorkers(shares) } : {}),
publishedAt: workOrder.published_at, publishedAt: workOrder.published_at,
pinnedAt: workOrder.pinned_at, pinnedAt: workOrder.pinned_at,
-6
View File
@@ -28,12 +28,6 @@ export type RuntimeConfig = {
maxFiles?: number maxFiles?: number
} }
retention: { retention: {
rawPayloadDays: number
taskEventDays: number
webhookEventDays: number
auditLogDays: number
/** 工单事件(work_order_events)保留天数:这张表只增不减,必须纳入清理。 */
workOrderEventDays: number
/** 已处理成本异常(cost_exceptions)保留天数:超期归档,open 行不归档。 */ /** 已处理成本异常(cost_exceptions)保留天数:超期归档,open 行不归档。 */
costExceptionDays: number costExceptionDays: number
} }
@@ -782,6 +782,8 @@ export type WorkOrder = {
username: string username: string
displayName: string displayName: string
} }
/** 寄售供货打手 id:后台「接单人员」筛选同时认这个 id。 */
consignmentWorkerId?: number | null
/** 拼单订单的当前参与打手,普通整单不返回该字段。 */ /** 拼单订单的当前参与打手,普通整单不返回该字段。 */
sharingWorkers?: Array<{ sharingWorkers?: Array<{
workerId: number workerId: number
@@ -155,6 +155,18 @@ export function isExplicitOrderNoSearch(text: unknown): boolean {
) )
} }
/**
* 快照里是否带寄售履约标记。
*
* 寄售单的供货打手不在 worker / sharingWorkers 上,而在库存行上;
* 旧快照没有 consignmentWorkerId 时无法判断它是否命中「接单人员」筛选。
*/
function hasConsignmentMaterial(snapshot: Record<string, unknown>): boolean {
const material = isRecord(snapshot.material) ? snapshot.material : null
const consignment = material && isRecord(material.consignment) ? material.consignment : null
return Boolean(consignment && (consignment.inventoryId || consignment.inventoryNo))
}
export function evaluateAdminWorkOrderRealtimeMatch( export function evaluateAdminWorkOrderRealtimeMatch(
snapshot: unknown, snapshot: unknown,
filters: AdminWorkOrderListFilters, filters: AdminWorkOrderListFilters,
@@ -190,7 +202,16 @@ export function evaluateAdminWorkOrderRealtimeMatch(
toNumber((worker as { workerId?: unknown })?.workerId), toNumber((worker as { workerId?: unknown })?.workerId),
) )
: [] : []
if (assignedWorkerId !== filters.workerId && !sharingWorkerIds.includes(filters.workerId)) { // 后台「接单人员」筛选同时认寄售供货人;实时快照带 consignmentWorkerId。
const consignmentWorkerId = toNumber(snapshot.consignmentWorkerId)
const matchesTargetWorker =
assignedWorkerId === filters.workerId ||
sharingWorkerIds.includes(filters.workerId) ||
consignmentWorkerId === filters.workerId
if (!matchesTargetWorker) {
// 寄售单的供货人只存在库存行上:旧快照 / 未加载库存时无法证伪,
// 直接删行会把「接单人员 + 内部寄售匹配」列表里的寄售单清空。
if (consignmentWorkerId <= 0 && hasConsignmentMaterial(snapshot)) return 'unknown'
return 'mismatch' return 'mismatch'
} }
} }
@@ -332,11 +332,27 @@ test('打手 / 分类 / 工单号 / 创建日期筛选的证伪能力', () => {
}), }),
)! )!
// 指派给别的打手 → mismatch // 指派给别的打手 → mismatch(consignmentWorkerId 明确指向别人,才能证伪)
assert.equal( assert.equal(
evaluateAdminWorkOrderRealtimeMatch(buildSnapshot({ worker: { workerId: 8 } }), filters), evaluateAdminWorkOrderRealtimeMatch(
buildSnapshot({ worker: { workerId: 8 }, consignmentWorkerId: 8 }),
filters,
),
'mismatch', 'mismatch',
) )
// 寄售供货人是目标打手 → 命中(内部寄售直接履约没有 assigned_worker_id)
assert.equal(
evaluateAdminWorkOrderRealtimeMatch(
buildSnapshot({ consignmentWorkerId: 7, categoryId: 12 }),
filters,
),
'match',
)
// 旧快照没有 consignmentWorkerId、但有寄售标记 → 无法证伪,只能原地更新,不能删行
assert.equal(
evaluateAdminWorkOrderRealtimeMatch(buildSnapshot({ categoryId: 12 }), filters),
'unknown',
)
// 拼单参与者是目标打手 → 命中(后端 workerId 筛选也认拼单参与者) // 拼单参与者是目标打手 → 命中(后端 workerId 筛选也认拼单参与者)
assert.equal( assert.equal(
evaluateAdminWorkOrderRealtimeMatch( evaluateAdminWorkOrderRealtimeMatch(
@@ -3,16 +3,16 @@ type: 权威架构与实施计划
description: 订单、内部 SKU、履约路由、履约执行、商品、成本与财务的唯一权威口径及重构计划。 description: 订单、内部 SKU、履约路由、履约执行、商品、成本与财务的唯一权威口径及重构计划。
tags: [订单, 内部SKU, 履约, 商品, 成本, 财务, 迁移] tags: [订单, 内部SKU, 履约, 商品, 成本, 财务, 迁移]
status: authoritative status: authoritative
version: 2.4 version: 2.5
updated_at: 2026-11-08 updated_at: 2026-11-09
--- ---
# 统一订单、履约与财务模型 # 统一订单、履约与财务模型
> 状态:**唯一权威文档**。 > 状态:**唯一权威文档**。
> 最后校准:2026-11-08。 > 最后校准:2026-11-09。
> 生产基线:迁移 `146`。 > 生产基线:迁移 `151`。
> 开发基线:迁移 `150`(仅本地,尚未部署生产)。 > 开发基线:迁移 `152`(仅本地,尚未部署生产)。
> >
> 从本文件生效起,订单、内部 SKU、履约渠道、履约载体、标准商品、成本和财务相关实现, > 从本文件生效起,订单、内部 SKU、履约渠道、履约载体、标准商品、成本和财务相关实现,
> 只能以本文件为业务口径。其他旧分析只作为历史资料,不再具备设计或验收效力。 > 只能以本文件为业务口径。其他旧分析只作为历史资料,不再具备设计或验收效力。
@@ -313,7 +313,7 @@ finance_journal_* -- 总账凭证,可重建
| 信息 | 唯一写入方 | 唯一真源 | 明确禁止 | | 信息 | 唯一写入方 | 唯一真源 | 明确禁止 |
| --- | --- | --- | --- | | --- | --- | --- | --- |
| 平台订单与订单项 | 订单接入服务 | `orders` / `order_items` | 财务或履约服务改写平台事实 | | 平台订单与订单项 | 订单接入服务 | `orders` / `order_items` | 财务或履约服务改写平台事实 |
| 平台原始报文 | 各平台适配器 | 原始表 / `raw_payload_json` | 覆盖历史报文 | | 平台原始报文 | 各平台适配器 | `order_source_snapshots` / 首次 `raw_payload_json` | 覆盖历史报文 |
| 中文内部 SKU | 商品身份服务 | `internal_skus` | 各业务表各自发明同义 SKU | | 中文内部 SKU | 商品身份服务 | `internal_skus` | 各业务表各自发明同义 SKU |
| 平台字段到内部 SKU | 映射服务 | `internal_sku_mappings` | 按名称猜测渠道 | | 平台字段到内部 SKU | 映射服务 | `internal_sku_mappings` | 按名称猜测渠道 |
| 主渠道与兜底 | 路由服务 | `sku_channel_routes` | UI 根据商品前缀筛选 | | 主渠道与兜底 | 路由服务 | `sku_channel_routes` | UI 根据商品前缀筛选 |
@@ -366,6 +366,7 @@ finance_journal_* -- 总账凭证,可重建
- `orders` - `orders`
- `order_items` 的原始平台字段 - `order_items` 的原始平台字段
- 平台原始订单表与原始报文字段 - 平台原始订单表与原始报文字段
- `order_source_snapshots` 平台订单不可变事件快照
- 平台售后、退款、支付事实 - 平台售后、退款、支付事实
- `work_orders`、`work_order_shares`、`work_order_events` - `work_orders`、`work_order_shares`、`work_order_events`
- `fulfillment_tasks` 中真实执行与结果记录 - `fulfillment_tasks` 中真实执行与结果记录
@@ -377,6 +378,14 @@ finance_journal_* -- 总账凭证,可重建
A 级数据只能追加新事件或做字段兼容迁移,不能为了“模型整齐”重写历史事实。 A 级数据只能追加新事件或做字段兼容迁移,不能为了“模型整齐”重写历史事实。
订单重复同步必须遵守:
- 首次写入的 `orders.raw_payload_json` 不再覆盖;
- 每次来源同步追加 `order_source_snapshots`,相同内容幂等;
- 已有 `order_items` 的平台 SKU、名称、数量、规格和快照不因重推更新;
- 重推中缺少的订单项只记录为来源差异,不得物理删除订单项或级联删除履约记录;
- 数据保留任务不得清空或删除 A 级事实,只允许维护可重建的派生数据。
### B 级:业务输入,必须迁移而不是重建 ### B 级:业务输入,必须迁移而不是重建
- `internal_sku_mappings` - `internal_sku_mappings`