dachui/order_site
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

重构：硬切履约状态唯一真源

Browse Source
This commit is contained in:
yml2213
2026-11-08 19:44:27 +08:00
parent 6bff1971dd
commit bae9a5ae5e
54 changed files with 2096 additions and 437 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/scripts/audit-fulfillment-state.ts
+16 -15
View File
@@ -93,7 +93,7 @@ async function loadRecentDuplicateAttempts(): Promise<CountRow[]> {
return result.rows
}
async function loadShadowModelDiffs(): Promise<CountRow[]> {
async function loadTruthIntegrityChecks(): Promise<CountRow[]> {
const result = await query<{
missing_fulfillments: number
fulfillment_state_diffs: number
@@ -116,7 +116,7 @@ async function loadShadowModelDiffs(): Promise<CountRow[]> {
fulfillment.issue_category
) IS DISTINCT FROM ROW(
CASE WHEN state.route_status = 'configured' THEN 'configured' ELSE 'unrouted' END,
state.intended_channel,
state.stored_channel,
state.carrier_status,
state.current_attempt_type,
NULLIF(state.current_attempt_id, 0),
@@ -162,8 +162,6 @@ async function loadShadowModelDiffs(): Promise<CountRow[]> {
(
SELECT COUNT(*)::int
FROM fulfillment_state_events event
JOIN v_fulfillment_item_state active_state
ON active_state.order_item_id = event.order_item_id
WHERE event.fulfillment_id IS NULL
) AS unlinked_events
FROM v_fulfillment_item_state state
@@ -173,20 +171,20 @@ async function loadShadowModelDiffs(): Promise<CountRow[]> {
)
const row = result.rows[0]
return [
{ label: '影子履约缺失', count: Number(row?.missing_fulfillments || 0) },
{ label: '影子履约状态差异', count: Number(row?.fulfillment_state_diffs || 0) },
{ label: '影子执行尝试缺失', count: Number(row?.missing_attempts || 0) },
{ label: '影子开放异常缺失', count: Number(row?.missing_open_issues || 0) },
{ label: '履约事件未关联影子记录', count: Number(row?.unlinked_events || 0) },
{ label: '正式履约缺失', count: Number(row?.missing_fulfillments || 0) },
{ label: '正式履约状态差异', count: Number(row?.fulfillment_state_diffs || 0) },
{ label: '执行尝试缺失', count: Number(row?.missing_attempts || 0) },
{ label: '开放异常缺失', count: Number(row?.missing_open_issues || 0) },
{ label: '履约事件未关联正式记录', count: Number(row?.unlinked_events || 0) },
]
}
async function main(): Promise<void> {
const [counts, issues, duplicates, shadowDiffs] = await Promise.all([
const [counts, issues, duplicates, integrityChecks] = await Promise.all([
loadCounts(),
loadIssueBreakdown(),
loadRecentDuplicateAttempts(),
loadShadowModelDiffs(),
loadTruthIntegrityChecks(),
])
console.info('[fulfillment-audit] 总览')
@@ -208,20 +206,23 @@ async function main(): Promise<void> {
console.info(` ${row.label}: ${row.count}`)
}
console.info('[fulfillment-audit] 影子模型差异')
for (const row of shadowDiffs) {
console.info('[fulfillment-audit] 正式履约完整性')
for (const row of integrityChecks) {
console.info(` ${row.label}: ${row.count}`)
}
const critical = new Map(counts.map((row) => [row.label, Number(row.count || 0)]))
const duplicateCount = Number(duplicates[0]?.count || 0)
const shadowDiffCount = shadowDiffs.reduce((sum, row) => sum + Number(row.count || 0), 0)
const integrityDifferenceCount = integrityChecks.reduce(
(sum, row) => sum + Number(row.count || 0),
0,
)
const hasCritical =
(critical.get('状态不一致') || 0) > 0 ||
(critical.get('售后关闭仍进入履约队列') || 0) > 0 ||
(critical.get('已分配但无有效载体') || 0) > 0 ||
duplicateCount > 0 ||
shadowDiffCount > 0
integrityDifferenceCount > 0
if (strict && hasCritical) {
process.exitCode = 1
apps/backend/src/db/migrations/163_cutover_fulfillment_truth.sql
+1351
View File
File diff suppressed because it is too large Load Diff
apps/backend/src/db/migrations/164_fulfillment_event_cascade.sql
+13
View File
@@ -0,0 +1,13 @@
-- 164_fulfillment_event_cascade.sql
--
-- 履约事件已经由 fulfillment_id 直接归属正式履约记录。
-- 订单项删除时事件应随履约记录级联清理，不能先 SET NULL 再触发 append-only 更新保护。
ALTER TABLE fulfillment_state_events
DROP CONSTRAINT IF EXISTS fk_fulfillment_state_events_fulfillment;
ALTER TABLE fulfillment_state_events
ADD CONSTRAINT fk_fulfillment_state_events_fulfillment
FOREIGN KEY (fulfillment_id)
REFERENCES fulfillments(id)
ON DELETE CASCADE;
apps/backend/src/db/migrations/165_fulfillment_event_append_only_update.sql
+16
View File
@@ -0,0 +1,16 @@
-- 165_fulfillment_event_append_only_update.sql
--
-- 事件语义是“应用不可改”，父记录删除时允许数据库级联清理。
-- 旧环境里曾存在 BEFORE DELETE OR UPDATE 的漂移定义，这里统一收回为只阻止 UPDATE。
DROP TRIGGER IF EXISTS trg_fulfillment_state_events_append_only
ON fulfillment_state_events;
CREATE TRIGGER trg_fulfillment_state_events_append_only
BEFORE UPDATE ON fulfillment_state_events
FOR EACH ROW
EXECUTE FUNCTION reject_fulfillment_state_event_mutation();
COMMENT ON TRIGGER trg_fulfillment_state_events_append_only
ON fulfillment_state_events IS
'禁止应用修改履约事件；数据库外键级联删除不受影响';
apps/backend/src/db/migrations/166_fulfillment_insert_resolved_sku.sql
+59
View File
@@ -0,0 +1,59 @@
-- 166_fulfillment_insert_resolved_sku.sql
--
-- resolve_internal_sku 可能返回 NULL。正式履约行的 internal_sku 是不带 NULL 的稳定键，
-- 无法解析时统一落空字符串，并立即进入 missing_internal_sku 异常口径。
CREATE OR REPLACE FUNCTION ensure_fulfillment_for_order_item()
RETURNS TRIGGER
LANGUAGE plpgsql
AS $$
BEGIN
INSERT INTO fulfillments (
order_item_id,
order_id,
internal_sku,
route_status,
channel,
carrier_status,
current_attempt_type,
current_attempt_source_id,
current_attempt_status,
lock_reason,
issue_code,
issue_category,
retryable,
owner_role,
created_at,
updated_at,
status,
source,
locked,
channel_switch_reason
) VALUES (
NEW.id,
NEW.order_id,
COALESCE(resolve_internal_sku(NEW.manual_sku_key, NEW.sku_name, NEW.sku_code), ''),
'unrouted',
NULL,
'missing',
'',
NULL,
'',
'',
'',
'',
FALSE,
'',
NEW.created_at,
NOW(),
'pending',
'',
FALSE,
''
)
ON CONFLICT (order_item_id) DO NOTHING;
PERFORM refresh_fulfillment_derived(NEW.id);
RETURN NEW;
END;
$$;
apps/backend/src/repositories/kuaishou-industry-voucher-repo.ts
+2 -2
View File
@@ -355,10 +355,10 @@ export async function listKuaishouIndustryVouchersForAdmin(
COALESCE(ft.task_no, '') AS task_no
FROM kuaishou_industry_vouchers v
LEFT JOIN fulfillment_tasks ft ON ft.id = v.task_id
LEFT JOIN order_items oi_task ON oi_task.id = ft.order_item_id
LEFT JOIN v_order_item_current oi_task ON oi_task.id = ft.order_item_id
LEFT JOIN LATERAL (
SELECT oi.sku_code, oi.sku_name
FROM order_items oi
FROM v_order_item_current oi
WHERE v.order_id IS NOT NULL
AND oi.order_id = v.order_id
ORDER BY oi.id ASC
apps/backend/src/repositories/order-item-repo.test.ts
+1 -1
View File
@@ -186,7 +186,7 @@ test('已分配订单重推写库时不会用平台新 SKU 覆盖冻结值', asy
let selectCount = 0
const executor = async (sql: string, params: unknown[] = []) => {
if (sql.includes('SELECT *') && sql.includes('FROM order_items')) {
if (sql.includes('SELECT *') && sql.includes('FROM v_order_item_current')) {
selectCount += 1
return { rows: selectCount === 1 ? [existing] : [] }
}
apps/backend/src/repositories/order-item-repo.ts
+17 -8
View File
@@ -29,7 +29,7 @@ export async function listOrderItemsByOrderIds(orderIds: number[]): Promise<Orde
const result = await query<OrderItemRow>(
`
SELECT *
FROM order_items
FROM v_order_item_current
WHERE order_id = ANY($1::bigint[])
ORDER BY order_id ASC, id ASC
`,
@@ -316,7 +316,7 @@ async function listDeletableOrderItemIdsWithExecutor(
const result = await executor(
`
SELECT oi.id
FROM order_items oi
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
@@ -336,7 +336,7 @@ async function listOrderItemsByOrderIdWithExecutor(
const result = await executor(
`
SELECT *
FROM order_items
FROM v_order_item_current
WHERE order_id = $1
ORDER BY id ASC
`,
@@ -347,9 +347,10 @@ async function listOrderItemsByOrderIdWithExecutor(
}
export async function getOrderItemById(orderItemId: number | string): Promise<OrderItemRow | null> {
const result = await query<OrderItemRow>('SELECT * FROM order_items WHERE id = $1 LIMIT 1', [
Number(orderItemId),
])
const result = await query<OrderItemRow>(
'SELECT * FROM v_order_item_current WHERE id = $1 LIMIT 1',
[Number(orderItemId)],
)
return result.rows[0] || null
}
@@ -358,9 +359,17 @@ export async function updateOrderItemSourceSnapshot(
input: { skuName: string; itemSnapshotJson: string; updatedAt: string; specJson?: string },
): Promise<OrderItemRow | null> {
return withTransaction(async (client) => {
const id = Number(orderItemId)
const orderItemLock = await client.query<{ id: number }>(
`SELECT id FROM order_items WHERE id = $1 FOR UPDATE`,
[id],
)
if (!orderItemLock.rows[0]) return null
await client.query(`SELECT id FROM fulfillments WHERE order_item_id = $1 FOR UPDATE`, [id])
const selected = await client.query<OrderItemRow>(
`SELECT * FROM order_items WHERE id = $1 FOR UPDATE`,
[Number(orderItemId)],
`SELECT * FROM v_order_item_current WHERE id = $1`,
[id],
)
const existing = selected.rows[0]
if (!existing) return null
apps/backend/src/repositories/order-repo.test.ts
+5 -5
View File
@@ -107,8 +107,8 @@ test('订单列表按当前履约模式筛选，不再依赖任务执行器', as
channel: 'manual',
})
assert.match(calls[0]?.sql || '', /oi\.fulfillment_channel = \$1/)
assert.match(calls[0]?.sql || '', /oi\.fulfillment_channel IS DISTINCT FROM \$1/)
assert.match(calls[0]?.sql || '', /fulfillment\.channel = \$1/)
assert.match(calls[0]?.sql || '', /fulfillment\.channel IS DISTINCT FROM \$1/)
assert.doesNotMatch(calls[0]?.sql || '', /COUNT\(DISTINCT/)
assert.doesNotMatch(calls[0]?.sql || '', /ft\.executor_key/)
assert.deepEqual(calls[0]?.params, ['manual'])
@@ -122,7 +122,7 @@ test('订单列表待判定筛选不扫描任务执行器', async () => {
})
assert.match(calls[0]?.sql || '', /EXISTS/)
assert.match(calls[0]?.sql || '', /oi\.fulfillment_status = 'pending'/)
assert.match(calls[0]?.sql || '', /fulfillment\.status = 'pending'/)
assert.doesNotMatch(calls[0]?.sql || '', /COUNT\(DISTINCT/)
assert.deepEqual(calls[0]?.params, [])
})
@@ -134,7 +134,7 @@ test('订单列表历史平台筛选映射为 archived', async () => {
channel: 'legacy_platform',
})
assert.match(calls[0]?.sql || '', /oi\.fulfillment_status = 'archived'/)
assert.match(calls[0]?.sql || '', /fulfillment\.status = 'archived'/)
assert.deepEqual(calls[0]?.params, [])
})
@@ -145,7 +145,7 @@ test('订单列表混合履约筛选按订单项聚合判定', async () => {
channel: 'mixed',
})
assert.match(calls[0]?.sql || '', /HAVING COUNT\(DISTINCT oi\.fulfillment_channel\) > 1/)
assert.match(calls[0]?.sql || '', /HAVING COUNT\(DISTINCT fulfillment\.channel\) > 1/)
assert.deepEqual(calls[0]?.params, [])
})
apps/backend/src/repositories/order-repo.ts
+18 -18
View File
@@ -279,7 +279,7 @@ export async function listOrdersWithExecutor(
) ILIKE ${keywordParam}
OR EXISTS (
SELECT 1
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.order_id = o.id
AND (
(
@@ -339,36 +339,36 @@ export async function listOrdersWithExecutor(
const normalizedMode = channel === 'legacy_platform' ? 'archived' : channel
if (normalizedMode === 'pending') {
filters.push(`EXISTS (
SELECT 1 FROM order_items oi
WHERE oi.order_id = o.id AND oi.fulfillment_status = 'pending'
SELECT 1 FROM fulfillments fulfillment
WHERE fulfillment.order_id = o.id AND fulfillment.status = 'pending'
)`)
} else if (normalizedMode === 'archived') {
filters.push(`EXISTS (
SELECT 1 FROM order_items oi
WHERE oi.order_id = o.id AND oi.fulfillment_status = 'archived'
SELECT 1 FROM fulfillments fulfillment
WHERE fulfillment.order_id = o.id AND fulfillment.status = 'archived'
)`)
} else if (normalizedMode === 'mixed') {
filters.push(`o.id IN (
SELECT oi.order_id FROM order_items oi
WHERE oi.fulfillment_status = 'assigned'
GROUP BY oi.order_id
HAVING COUNT(DISTINCT oi.fulfillment_channel) > 1
SELECT fulfillment.order_id FROM fulfillments fulfillment
WHERE fulfillment.status = 'assigned'
GROUP BY fulfillment.order_id
HAVING COUNT(DISTINCT fulfillment.channel) > 1
)`)
} else {
params.push(normalizedMode)
const modeParam = `$${params.length}`
filters.push(`(
EXISTS (
SELECT 1 FROM order_items oi
WHERE oi.order_id = o.id
AND oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = ${modeParam}
SELECT 1 FROM fulfillments fulfillment
WHERE fulfillment.order_id = o.id
AND fulfillment.status = 'assigned'
AND fulfillment.channel = ${modeParam}
)
AND NOT EXISTS (
SELECT 1 FROM order_items oi
WHERE oi.order_id = o.id
AND oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel IS DISTINCT FROM ${modeParam}
SELECT 1 FROM fulfillments fulfillment
WHERE fulfillment.order_id = o.id
AND fulfillment.status = 'assigned'
AND fulfillment.channel IS DISTINCT FROM ${modeParam}
)
)`)
}
@@ -377,7 +377,7 @@ export async function listOrdersWithExecutor(
if (skuCode) {
params.push(skuCode)
filters.push(
`EXISTS (SELECT 1 FROM order_items oi WHERE oi.order_id = o.id AND oi.sku_code = $${params.length})`,
`EXISTS (SELECT 1 FROM v_order_item_current oi WHERE oi.order_id = o.id AND oi.sku_code = $${params.length})`,
)
}
apps/backend/src/repositories/redeem-certificate-repo.ts
+2 -2
View File
@@ -167,10 +167,10 @@ export async function listRedeemCertificatesForAdmin(
ORDER BY ft.id ASC
LIMIT 1
) ft ON true
LEFT JOIN order_items oi_task ON oi_task.id = ft.order_item_id
LEFT JOIN v_order_item_current oi_task ON oi_task.id = ft.order_item_id
LEFT JOIN LATERAL (
SELECT oi.sku_code, oi.sku_name
FROM order_items oi
FROM v_order_item_current oi
WHERE c.local_order_id IS NOT NULL
AND oi.order_id = c.local_order_id
ORDER BY oi.id ASC
apps/backend/src/repositories/task-repo.ts
+5 -3
View File
@@ -96,7 +96,9 @@ export async function createTask(input: TaskCreateInput): Promise<TaskRow | null
return withTransaction(async (client) => {
// 同一订单项单元的创建必须串行，否则并发调度会同时算出同一尝试号，
// 再一起绕过“仅一个当前尝试”的唯一索引。
await client.query(`SELECT id FROM order_items WHERE id = $1 FOR UPDATE`, [input.orderItemId])
await client.query(`SELECT id FROM fulfillments WHERE order_item_id = $1 FOR UPDATE`, [
input.orderItemId,
])
const active = await client.query<{ id: number }>(
`SELECT id
FROM fulfillment_tasks
@@ -483,7 +485,7 @@ export async function listTasks({
const itemsParams = [...params, pageSize, offset]
const itemsResult = await query<TaskRow>(
`${buildTaskSelect('oi.sku_code, oi.sku_name, oi.quantity')}
LEFT JOIN order_items oi ON oi.id = ft.order_item_id
LEFT JOIN v_order_item_current oi ON oi.id = ft.order_item_id
${whereClause}
ORDER BY ft.id DESC
LIMIT $${itemsParams.length - 1} OFFSET $${itemsParams.length}`,
@@ -504,7 +506,7 @@ export async function listTasks({
const needsJoinKcts = whereClause.includes('kcts.')
const joinClauses: string[] = []
if (needsJoinOi) {
joinClauses.push('LEFT JOIN order_items oi ON oi.id = ft.order_item_id')
joinClauses.push('LEFT JOIN v_order_item_current oi ON oi.id = ft.order_item_id')
}
if (needsJoinCtx) {
joinClauses.push('LEFT JOIN task_runtime_contexts ctx ON ctx.task_id = ft.id')
apps/backend/src/services/admin/admin-order-read-helpers.ts
+1 -1
View File
@@ -202,7 +202,7 @@ export function resolveVoucherSummary(vouchers: KuaishouIndustryVoucherRow[] = [
}
/**
* 订单级履约摘要：只读 order_items.fulfillment_channel / fulfillment_status。
* 订单级履约摘要：只读 fulfillments.channel / status 的订单项投影。
* 多商品订单出现不同模式时显示 mixed；尚未判定的订单项单独计数。
*/
export function resolveOrderFulfillmentSummary({
apps/backend/src/services/cost-center/affiliate-product-service.ts
+6 -6
View File
@@ -23,7 +23,7 @@ export async function cleanupDirtyAffiliateDashProducts() {
oi.id AS order_item_id,
clean.id AS clean_product_id,
dirty.id AS expected_product_id
FROM order_items oi
FROM v_order_item_current oi
JOIN product_catalog dirty ON dirty.id = oi.canonical_product_id
JOIN fulfillment_tasks ft ON ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
JOIN product_catalog clean ON clean.canonical_sku = 'aff:' || COALESCE(
@@ -58,7 +58,7 @@ export async function cleanupDirtyAffiliateDashProducts() {
await query(`
DELETE FROM product_catalog
WHERE canonical_sku ~ '^aff:[^a-z0-9_]'
AND NOT EXISTS (SELECT 1 FROM order_items WHERE canonical_product_id = product_catalog.id)
AND NOT EXISTS (SELECT 1 FROM fulfillments WHERE canonical_product_id = product_catalog.id)
AND NOT EXISTS (SELECT 1 FROM cost_versions WHERE canonical_product_id = product_catalog.id)
AND NOT EXISTS (SELECT 1 FROM cost_snapshots WHERE canonical_product_id = product_catalog.id)
AND NOT EXISTS (SELECT 1 FROM sales_daily_facts WHERE canonical_product_id = product_catalog.id)
@@ -207,7 +207,7 @@ export async function backfillAdminAffiliateDashSkus() {
FROM affiliate_item ai
JOIN product_catalog p
ON p.canonical_sku = 'aff:' || ai.sku_code
JOIN order_items oi ON oi.id = ai.id
JOIN v_order_item_current oi ON oi.id = ai.id
)
SELECT id, canonical_product_id, expected_product_id
FROM target
@@ -240,7 +240,7 @@ export async function backfillAdminAffiliateDashSkus() {
'YYYY-MM-DD'
)
FROM affiliate_item
JOIN order_items oi ON oi.id = affiliate_item.id
JOIN v_order_item_current oi ON oi.id = affiliate_item.id
JOIN orders o ON o.id = oi.order_id
) AS from_date,
(
@@ -249,7 +249,7 @@ export async function backfillAdminAffiliateDashSkus() {
'YYYY-MM-DD'
)
FROM affiliate_item
JOIN order_items oi ON oi.id = affiliate_item.id
JOIN v_order_item_current oi ON oi.id = affiliate_item.id
JOIN orders o ON o.id = oi.order_id
) AS to_date
`,
@@ -275,7 +275,7 @@ export async function backfillAdminAffiliateDashSkus() {
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.canonical_product_id IS NULL
AND EXISTS (
SELECT 1 FROM fulfillment_tasks ft
apps/backend/src/services/cost-center/cost-center-constants.ts
+1 -1
View File
@@ -58,7 +58,7 @@ export const COST_MODEL_CODE_BASE: Record<string, string> = {
other: 'other_cost',
}
/** 履约渠道：唯一真源是 order_items.fulfillment_channel + fulfillment_status。 */
/** 履约渠道：唯一真源是 fulfillments.channel + status。 */
export type FulfillmentChannel =
| 'work_order'
| 'affiliate_dash'
apps/backend/src/services/cost-center/cost-center-overview-service.ts
+2 -2
View File
@@ -80,7 +80,7 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
(
SELECT COUNT(*)::int
FROM cost_exceptions ce
JOIN order_items oi ON oi.id = ce.order_item_id
JOIN v_order_item_current oi ON oi.id = ce.order_item_id
WHERE ce.status = 'open'
AND (
$1::text IS NULL
@@ -89,7 +89,7 @@ export async function getAdminCostCenterOverview(params: { scope?: string } = {}
) AS open_exception_count,
(
SELECT COUNT(*)::int
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.canonical_product_id IS NULL
-- 只统计已定渠道的单：待分类单还没有渠道，不能算「未映射」。
AND oi.fulfillment_status = 'assigned'
apps/backend/src/services/cost-center/cost-center-shared.ts
+1 -1
View File
@@ -2,7 +2,7 @@
* 成本中心共享层入口。
*
* 口径（2026-10-23 重写）：
* 1. 履约渠道只有 order_items.fulfillment_channel/status 一个真源；
* 1. 履约渠道只有 fulfillments.channel/status 一个真源；
* 2. 成本项用 cost_model 表达「成本怎么算」，与履约方式正交；
* 3. 商品来源由 canonical_sku 前缀推导，商品表不再存履约载体；
* 4. 快照与异常表都不再存履约载体，join order_items 即可拿到。
apps/backend/src/services/cost-center/cost-center.test.ts
+1 -1
View File
@@ -332,7 +332,7 @@ test('成本中心总览异常统计必须走异常表关联，不能全表扫
new URL('./cost-center-overview-service.ts', import.meta.url),
'utf8',
)
assert.ok(source.includes('JOIN order_items oi ON oi.id = ce.order_item_id'))
assert.ok(source.includes('JOIN v_order_item_current oi ON oi.id = ce.order_item_id'))
assert.ok(!source.includes('ce.order_item_id IN ('))
assert.ok(source.includes('productChannelFilterSql(1)'))
})
apps/backend/src/services/cost-center/cost-definition-service.ts
+4 -4
View File
@@ -343,7 +343,7 @@ export async function listAdminCostVersions(payload: JsonObject = {}) {
ROUND(AVG(wo.reward_amount))::bigint AS reward_avg_fen,
SUM(wo.reward_amount)::bigint AS reward_fen
FROM work_orders wo
JOIN order_items oi ON oi.id = wo.order_item_id
JOIN v_order_item_current oi ON oi.id = wo.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE o.pay_status IN ('paid', 'refunded')
GROUP BY wo.product_rule_id
@@ -583,7 +583,7 @@ export async function listAdminCostExceptions(payload: JsonObject = {}) {
params.push(channel)
clauses.push(
`ce.order_item_id IN (
SELECT oi.id FROM order_items oi WHERE COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) = $${params.length}
SELECT oi.id FROM v_order_item_current oi WHERE COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) = $${params.length}
)`,
)
}
@@ -600,7 +600,7 @@ export async function listAdminCostExceptions(payload: JsonObject = {}) {
ce.*,
COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) AS channel
FROM cost_exceptions ce
LEFT JOIN order_items oi ON oi.id = ce.order_item_id
LEFT JOIN v_order_item_current oi ON oi.id = ce.order_item_id
${where}
ORDER BY ce.status ASC, ce.created_at DESC, ce.id DESC
LIMIT $${listParams.length - 1} OFFSET $${listParams.length}
@@ -642,7 +642,7 @@ export async function summarizeAdminCostExceptions(
params.push(channel)
clauses.push(
`ce.order_item_id IN (
SELECT oi.id FROM order_items oi WHERE COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) = $${params.length}
SELECT oi.id FROM v_order_item_current oi WHERE COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) = $${params.length}
)`,
)
}
apps/backend/src/services/cost-center/cost-resolution-service.ts
+3 -3
View File
@@ -129,7 +129,7 @@ async function resolveManualCostHint(orderItemId: number | null) {
cfg.credit_quantity,
cfg.cost_mode,
COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS unit_price_fen
FROM order_items oi
FROM v_order_item_current oi
JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
WHERE oi.id = $1
@@ -317,7 +317,7 @@ async function resolveCostContext(payload: JsonObject): Promise<OrderItemCostCon
wo.id AS work_order_id,
wo.product_rule_id AS work_product_rule_id,
COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) AS channel
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN LATERAL (
SELECT id, product_rule_id
@@ -394,7 +394,7 @@ async function resolveCostContext(payload: JsonObject): Promise<OrderItemCostCon
}
/**
* 履约渠道只读 order_items.fulfillment_channel / fulfillment_status。
* 履约渠道只读 fulfillments.channel / status。
* 订单项还没判定时按人工处理，不允许用 provider 或核销码兜底。
*/
function resolveFulfillmentMode(value: unknown): FulfillmentChannel {
apps/backend/src/services/cost-center/cost-snapshot-service.ts
+1 -1
View File
@@ -317,7 +317,7 @@ const CANDIDATE_SQL = `
wpr.pricing_mode AS work_product_rule_pricing_mode,
wo.status AS work_order_status,
wo.reward_amount
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
apps/backend/src/services/cost-center/manual-fulfillment-config-service.ts
+7 -7
View File
@@ -156,7 +156,7 @@ export async function scanManualSkus(options: ScanOptions) {
COALESCE(oi.revenue_amount_fen, 0)::bigint AS revenue_fen,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date AS stat_date,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
@@ -273,7 +273,7 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
if (provider) {
params.push(provider)
conditions.push(`EXISTS (
SELECT 1 FROM order_items oi
SELECT 1 FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
AND o.provider = $${params.length}
@@ -314,7 +314,7 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
AS first_seen_date,
MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)
AS last_seen_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
@@ -381,7 +381,7 @@ export async function getManualSkuOverview() {
COUNT(*)::int AS item_count,
MAX((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)
AS last_seen_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
@@ -602,7 +602,7 @@ async function retryPendingManualItems(skuKey: string): Promise<number> {
if (!key) return 0
const candidates = await query<{ id: number }>(
`SELECT oi.id
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.fulfillment_status = 'pending'
AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = $1
ORDER BY oi.id
@@ -688,7 +688,7 @@ export async function deleteManualSkuClassification(
oi.id,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
AS basis_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
@@ -867,7 +867,7 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
oi.id AS order_item_id,
cfg.id AS config_id,
cfg.canonical_product_id
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
apps/backend/src/services/cost-center/product-catalog-service.ts
+13 -13
View File
@@ -410,7 +410,7 @@ export async function deleteAdminProduct(productId: number | string) {
(SELECT COUNT(*)::int FROM channel_product_mappings WHERE canonical_product_id = $1) AS mapping_count,
(SELECT COUNT(*)::int FROM cost_versions WHERE canonical_product_id = $1) AS version_count,
(SELECT COUNT(*)::int FROM cost_snapshots WHERE canonical_product_id = $1) AS snapshot_count,
(SELECT COUNT(*)::int FROM order_items WHERE canonical_product_id = $1) AS order_item_count
(SELECT COUNT(*)::int FROM fulfillments WHERE canonical_product_id = $1) AS order_item_count
`,
[id],
)
@@ -474,7 +474,7 @@ export async function deleteAdminVoucherProduct(productId: number | string) {
)
)
AS snapshot_count,
(SELECT COUNT(*)::int FROM order_items WHERE canonical_product_id = $1)
(SELECT COUNT(*)::int FROM fulfillments WHERE canonical_product_id = $1)
AS order_item_count
`,
[id],
@@ -779,7 +779,7 @@ export async function getAdminChannelMappingDiagnostics(payload: JsonObject = {}
COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) AS channel,
COUNT(*)::int AS item_count
FROM cost_exceptions ce
JOIN order_items oi ON oi.id = ce.order_item_id
JOIN v_order_item_current oi ON oi.id = ce.order_item_id
WHERE ce.reason = 'product_mapping_missing'
AND ce.status = 'open'
AND COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) = $1
@@ -874,7 +874,7 @@ export async function listAdminUnmappedMappingGroups(payload: JsonObject = {}) {
COALESCE(ce.detail_json->>'skuCode', '') AS sku_code,
COUNT(*)::int AS item_count
FROM cost_exceptions ce
JOIN order_items oi ON oi.id = ce.order_item_id
JOIN v_order_item_current oi ON oi.id = ce.order_item_id
WHERE ${clauses.join(' AND ')}
AND COALESCE(oi.fulfillment_channel, CASE WHEN oi.fulfillment_status = 'archived' THEN 'archived' ELSE 'pending' END) = $${totalParams.length}
GROUP BY 1, 2, 3
@@ -927,7 +927,7 @@ export async function listAdminUnmappedMappingGroups(payload: JsonObject = {}) {
MAX(NULLIF(ce.detail_json->>'basisAt', '')) AS last_seen_at,
COUNT(*)::int AS item_count
FROM cost_exceptions ce
JOIN order_items oi ON oi.id = ce.order_item_id
JOIN v_order_item_current oi ON oi.id = ce.order_item_id
WHERE ${clauses.join(' AND ')}
${modeClause}
GROUP BY 1, 2, 3, 4
@@ -1192,7 +1192,7 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) {
oi.sku_code,
oi.normalized_sku_name,
oi.normalized_sku_series
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND oi.id > $1
@@ -1241,7 +1241,7 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) {
AND ce.subject_type = 'order_item'
AND ce.reason = 'product_mapping_missing'
AND EXISTS (
SELECT 1 FROM order_items oi
SELECT 1 FROM v_order_item_current oi
WHERE oi.id = ce.subject_id AND oi.canonical_product_id IS NOT NULL
)
`,
@@ -1252,7 +1252,7 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) {
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND ($1 = '' OR o.provider = $1)
@@ -1354,7 +1354,7 @@ export async function backfillAdminNormalizedQuantities(payload: JsonObject = {}
`
SELECT oi.id, oi.sku_name, oi.sku_code, oi.normalized_sku_name,
oi.normalized_sku_series, oi.normalized_quantity
FROM order_items oi
FROM v_order_item_current oi
${range ? 'JOIN orders o ON o.id = oi.order_id' : ''}
WHERE (oi.normalized_quantity IS NULL
OR oi.normalized_sku_name IS NULL
@@ -1404,7 +1404,7 @@ export async function backfillAdminNormalizedQuantities(payload: JsonObject = {}
const remainingResult = await query<{ total: number }>(
`SELECT COUNT(*)::int AS total
FROM order_items oi
FROM v_order_item_current oi
${range ? 'JOIN orders o ON o.id = oi.order_id' : ''}
WHERE (oi.normalized_quantity IS NULL
OR oi.normalized_sku_name IS NULL
@@ -1758,7 +1758,7 @@ export async function setAdminProductPrice(
const earliest = await query<{ basis_date: string | null }>(
`
SELECT MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)::text AS basis_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id = $1
AND oi.fulfillment_status = 'assigned'
@@ -1840,7 +1840,7 @@ export async function setAdminProductPrice(
const missingItems = await query<{ id: number }>(
`
SELECT oi.id
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.canonical_product_id = $1
AND oi.fulfillment_status = 'assigned'
AND NOT EXISTS (
@@ -1866,7 +1866,7 @@ export async function setAdminProductPrice(
const coveredItems = await query<{ id: number }>(
`
SELECT oi.id
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.canonical_product_id = $1
AND oi.fulfillment_status = 'assigned'
AND EXISTS (
apps/backend/src/services/cost-center/product-normalize-service.ts
+3 -3
View File
@@ -11,7 +11,7 @@ import {
/**
* 管道 B：商品归一。
*
* 只写 `order_items.canonical_product_id`，不改渠道、不碰成本；
* 只写 `fulfillments.canonical_product_id`，不改渠道、不碰成本；
* 例外：三方卡券缺商品时按内部中文 SKU 自动建 `tpv:` 商品并补渠道规则。
*
* 解析顺序：
@@ -339,7 +339,7 @@ export async function normalizeOrderItemProducts(
wpr.rule_key AS work_rule_key,
cfg.canonical_product_id AS manual_config_product_id,
cfg_product.canonical_sku AS manual_config_product_sku
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
@@ -529,7 +529,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
wpr.rule_key AS work_rule_key,
cfg.canonical_product_id AS manual_config_product_id,
cfg_product.canonical_sku AS manual_config_product_sku
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
apps/backend/src/services/cost-center/template-bootstrap-service.ts
+3 -3
View File
@@ -626,7 +626,7 @@ async function resetImportedTemplateProductBootstrap(range: { fromDate: string;
`
WITH candidates AS (
SELECT oi.id, oi.canonical_product_id AS expected_product_id
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN channel_product_mappings m
ON m.provider = o.provider
@@ -680,7 +680,7 @@ async function resetImportedTemplateProductBootstrap(range: { fromDate: string;
AND m.note LIKE '接单模板 %自动初始化%'
AND EXISTS (
SELECT 1
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id = m.canonical_product_id
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
@@ -732,7 +732,7 @@ async function loadTemplateProductBootstrapData(range: { fromDate: string; toDat
wo.product_rule_id,
r.rule_key,
COUNT(DISTINCT oi.id)::int AS item_count
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
JOIN work_product_rules r ON r.id = wo.product_rule_id AND r.enabled = TRUE
apps/backend/src/services/cost-center/voucher-product-service.test.ts
+2 -1
View File
@@ -8,7 +8,8 @@ test('卡券 SKU 一个入口完成商品、规则、成本版本和订单回填
assert.ok(service.includes('withTransaction('))
assert.ok(service.includes("channel: 'third_party_voucher'"))
assert.ok(service.includes('sys_third_party_voucher_cost'))
assert.ok(service.includes('UPDATE order_items oi'))
assert.ok(service.includes('attachFulfillmentConfigurationsWithExecutor'))
assert.ok(!service.includes('UPDATE order_items oi'))
assert.ok(service.includes('syncCostSnapshotsForItems'))
assert.ok(service.includes('refreshFactsForDates'))
assert.ok(service.includes("resolveExceptionsForItemsByReason(result.itemIds, 'cost_missing')"))
apps/backend/src/services/cost-center/voucher-product-service.ts
+23 -15
View File
@@ -6,6 +6,7 @@ import type { AdminSession } from '../admin/admin-auth-service.js'
import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js'
import { upsertInternalSkuChannelWithClient } from '../fulfillment/internal-sku-channel-service.js'
import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js'
import { attachFulfillmentConfigurationsWithExecutor } from '../fulfillment/fulfillment-state-service.js'
import { resolvePriceEffectiveFrom } from './cost-center-normalizers.js'
import {
resolveExceptionsForItemsByReason,
@@ -103,24 +104,31 @@ export async function upsertAdminVoucherProduct(payload: JsonObject, session: Ad
replacePriority: true,
})
const repairedItems = await client.query<VoucherProductItemRow>(
const repairCandidates = await client.query<VoucherProductItemRow>(
`
UPDATE order_items oi
SET canonical_product_id = $2,
updated_at = $3
FROM orders o
WHERE o.id = oi.order_id
AND oi.fulfillment_channel = 'third_party_voucher'
AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = $1
AND oi.canonical_product_id IS DISTINCT FROM $2
RETURNING
SELECT
oi.id AS order_item_id,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
AS basis_date
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE TRUE
AND oi.fulfillment_channel = 'third_party_voucher'
AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = $1
AND oi.canonical_product_id IS DISTINCT FROM $2
`,
[skuKey, productId, now],
[skuKey, productId],
)
const itemIds = repairedItems.rows.map((row) => Number(row.order_item_id))
const itemIds = repairCandidates.rows.map((row) => Number(row.order_item_id))
if (itemIds.length > 0) {
await attachFulfillmentConfigurationsWithExecutor(client, {
mappings: itemIds.map((orderItemId) => ({
orderItemId,
canonicalProductId: productId,
})),
onlyWhenCanonicalProductDifferent: true,
})
}
if (itemIds.length > 0) {
await client.query(
`UPDATE cost_snapshots
@@ -145,13 +153,13 @@ export async function upsertAdminVoucherProduct(payload: JsonObject, session: Ad
}
let effectiveFrom = initialEffectiveFrom
if (!requestedEffectiveFrom && repairedItems.rows.length > 0) {
if (!requestedEffectiveFrom && repairCandidates.rows.length > 0) {
const missingDates = await client.query<{ basis_date: string }>(
`
SELECT MIN(
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
)::text AS basis_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = ANY($1::bigint[])
AND NOT EXISTS (
@@ -272,7 +280,7 @@ export async function upsertAdminVoucherProduct(payload: JsonObject, session: Ad
return {
product: mapProduct(product),
itemIds,
dates: [...new Set(repairedItems.rows.map((row) => row.basis_date).filter(Boolean))],
dates: [...new Set(repairCandidates.rows.map((row) => row.basis_date).filter(Boolean))],
costVersionId,
effectiveFrom,
amountFen,
apps/backend/src/services/cost-center/work-order-product-service.ts
+6 -6
View File
@@ -49,7 +49,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
'YYYY-MM-DD'
) AS to_date
FROM work_orders wo
JOIN order_items oi ON oi.id = wo.order_item_id
JOIN v_order_item_current oi ON oi.id = wo.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE wo.order_item_id IS NOT NULL
AND wo.status <> 'cancelled'
@@ -77,7 +77,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
wo.product_rule_id,
oi.canonical_product_id AS expected_product_id
FROM work_orders wo
JOIN order_items oi ON oi.id = wo.order_item_id
JOIN v_order_item_current oi ON oi.id = wo.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE wo.order_item_id IS NOT NULL
AND wo.status <> 'cancelled'
@@ -120,7 +120,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
SELECT
oi.normalized_sku_name AS name,
MIN(oi.sku_name) AS sample_name
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
@@ -174,7 +174,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
oi.sku_code,
MIN(oi.sku_name) AS sku_name,
oi.normalized_sku_name
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
@@ -219,7 +219,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
`
WITH tail AS (
SELECT oi.id, oi.normalized_sku_name
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND btrim(COALESCE(oi.normalized_sku_name, '')) <> ''
@@ -267,7 +267,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
const remainingResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id IS NULL
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
apps/backend/src/services/finance/customer-refund-service.ts
+2 -2
View File
@@ -411,10 +411,10 @@ async function loadRefundContext(workOrderIdValue: number | string): Promise<Ref
FROM order_refunds r
WHERE r.order_item_id = oi.id AND r.status = 'confirmed'
)), 0)
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.order_id = o.id
), 0)::bigint AS remaining_fen,
(SELECT COUNT(*)::int FROM order_items oi WHERE oi.order_id = o.id) AS item_count,
(SELECT COUNT(*)::int FROM v_order_item_current oi WHERE oi.order_id = o.id) AS item_count,
(SELECT COUNT(*)::int FROM work_orders w WHERE w.order_id = o.id) AS work_order_count
FROM work_orders wo
JOIN orders o ON o.id = wo.order_id
apps/backend/src/services/finance/finance-ledger-service.ts
+5 -4
View File
@@ -349,8 +349,9 @@ async function readSalesFactJournalTotals(client: PoolClient, statDate: string)
FROM finance_journal_entries e
JOIN finance_journal_lines l ON l.entry_id = e.id
WHERE e.ref_type = 'sales_fact'
AND e.ref_id = $1
AND e.status = 'posted'
AND e.occurred_at >= ($1::date::timestamp AT TIME ZONE 'Asia/Shanghai')
AND e.occurred_at < (($1::date + interval '1 day')::timestamp AT TIME ZONE 'Asia/Shanghai')
GROUP BY e.event_type, l.account_code
`,
[statDate],
@@ -612,7 +613,7 @@ async function assertPeriodCloseable(
const manualMissing = await client.query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
@@ -636,7 +637,7 @@ async function assertPeriodCloseable(
`
SELECT COUNT(*)::int AS total
FROM cost_snapshots cs
JOIN order_items oi ON oi.id = cs.order_item_id
JOIN v_order_item_current oi ON oi.id = cs.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE cs.subject_type = 'manual_case'
AND cs.superseded_at IS NULL AND reversed_at IS NULL
@@ -658,7 +659,7 @@ async function assertPeriodCloseable(
SELECT COUNT(*)::int AS total
FROM cost_snapshots cs
JOIN manual_expenses me ON me.order_item_id = cs.order_item_id
JOIN order_items oi ON oi.id = cs.order_item_id
JOIN v_order_item_current oi ON oi.id = cs.order_item_id
JOIN orders o ON o.id = oi.order_id
WHERE cs.subject_type = 'manual_case'
AND cs.superseded_at IS NULL AND reversed_at IS NULL
apps/backend/src/services/finance/finance-reconciliation-service.ts
+1 -1
View File
@@ -306,7 +306,7 @@ async function matchRow(
oi.id,
oi.order_id,
GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)::bigint AS revenue_fen
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = $3
AND o.provider = $1
apps/backend/src/services/finance/manual-fulfillment-service.ts
+6 -6
View File
@@ -242,7 +242,7 @@ export async function listManualFulfillmentPending(
const countResult = await query<{ total: number }>(
`${baseCte}
SELECT COUNT(*)::int AS total
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfilled ON fulfilled.order_item_id = oi.id
@@ -276,7 +276,7 @@ export async function listManualFulfillmentPending(
cfg.channels_json AS config_channels_json,
COALESCE((fp.config_json->>'rechargeUnitPriceFen')::bigint, 1) AS credit_unit_price_fen,
fulfilled.fulfilled_at
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
@@ -339,7 +339,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
wo.id AS work_order_id,
wo.work_order_no,
wo.product_rule_id AS work_product_rule_id
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfillment_profiles fp ON fp.profile_key = 'manual_' || cfg.platform_key
@@ -736,7 +736,7 @@ export async function refreshFactsForOrderItem(orderItemId: number): Promise<voi
`
SELECT (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
AS stat_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = $1
`,
@@ -838,7 +838,7 @@ async function matchManualFulfillmentConfigWithClient(client: PoolClient, orderI
}>(
`
SELECT cfg.id, cfg.canonical_product_id
FROM order_items oi
FROM v_order_item_current oi
JOIN manual_fulfillment_configs cfg ON cfg.enabled = TRUE
AND (
canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
@@ -1097,7 +1097,7 @@ export async function getManualFulfillmentCountsByOrderIds(
WHERE cfg.id IS NOT NULL AND fulfilled.fulfilled_at IS NULL
)::int AS pending_count,
COUNT(*) FILTER (WHERE cfg.id IS NULL)::int AS unconfigured_count
FROM order_items oi
FROM v_order_item_current oi
LEFT JOIN manual_fulfillment_configs cfg ON cfg.id = oi.fulfillment_config_id
LEFT JOIN fulfilled ON fulfilled.order_item_id = oi.id
WHERE oi.order_id = ANY($1::bigint[])
apps/backend/src/services/finance/manual-fulfillment.integration.test.ts
+39 -11
View File
@@ -30,13 +30,13 @@ test(
reopenFinancePeriod,
syncFinanceLedgerFromFacts,
} = await import('./finance-ledger-service.js')
const { rebuildSalesDailyFacts } = await import('./sales-facts-service.js')
const { rebuildChannelSalesDailyFacts } = await import('./sales-facts-channel-service.js')
const { importFinanceReconciliation } = await import('./finance-reconciliation-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const skuKey = `IT-MANUAL-${suffix}`
const creditSkuKey = `IT-CREDIT-${suffix}`
const unconfirmedSkuKey = `IT-PENDING-${suffix}`
const skuKey = `人工履约-${suffix}`
const creditSkuKey = `人工额度-${suffix}`
const unconfirmedSkuKey = `人工待确认-${suffix}`
const orderIds: number[] = []
const orderItemIds: number[] = []
const fundAccountIds: number[] = []
@@ -103,7 +103,7 @@ test(
})
assert.ok(rematched.updatedCount >= 1)
const matchedConfig = await query<{ fulfillment_config_id: number | null }>(
`SELECT fulfillment_config_id FROM order_items WHERE id = $1`,
`SELECT fulfillment_config_id FROM v_order_item_current WHERE id = $1`,
[cashItem.orderItemId],
)
assert.ok(Number(matchedConfig.rows[0]?.fulfillment_config_id) > 0, 'rematch 必须挂上配置')
@@ -375,12 +375,28 @@ test(
// 7. 总账：销售事实同步、试算平衡、关账阻断、平台对账匹配
const todayDate = shanghaiDate()
await rebuildSalesDailyFacts({ dateFrom: todayDate, dateTo: todayDate })
await rebuildChannelSalesDailyFacts({ dateFrom: todayDate, dateTo: todayDate })
const salesFacts = await query<{ count: number }>(
`SELECT COUNT(*)::int AS count
FROM sales_daily_facts
WHERE stat_date = $1::date`,
[todayDate],
)
assert.ok(Number(salesFacts.rows[0]?.count || 0) > 0, '必须先从统一履约投影生成销售事实')
const ledgerSync = await syncFinanceLedgerFromFacts({
dateFrom: todayDate,
dateTo: todayDate,
})
assert.ok(ledgerSync.created > 0, '销售事实必须生成总账凭证')
const journalEntries = await query<{ count: number }>(
`SELECT COUNT(*)::int AS count
FROM finance_journal_entries
WHERE period_key = $1`,
[todayDate.slice(0, 7)],
)
assert.ok(
ledgerSync.created > 0 || Number(journalEntries.rows[0]?.count || 0) > 0,
'销售事实必须生成总账凭证',
)
const trial = await getTrialBalance({ dateFrom: todayDate, dateTo: todayDate })
assert.equal(trial.balanced, true, '总账必须借贷平衡')
@@ -420,6 +436,8 @@ test(
)
// 清掉失败单后，关账才允许通过。
await confirmManualFulfillment({ orderItemId: failItem.orderItemId, costFen: 700 }, session)
await rebuildChannelSalesDailyFacts({ dateFrom: todayDate, dateTo: todayDate })
await syncFinanceLedgerFromFacts({ dateFrom: todayDate, dateTo: todayDate })
await closeFinancePeriod(periodKey, session)
await assert.rejects(
() =>
@@ -544,13 +562,23 @@ async function createManualItem(
const orderId = Number(order.rows[0]?.id)
const item = await query(
`INSERT INTO order_items (
order_id, sku_code, sku_name, channel, spec_json,
order_id, sku_code, sku_name, spec_json,
revenue_amount_fen, revenue_basis, created_at, updated_at
) VALUES ($1, $2, $3, 'manual', $4::jsonb, 5000, 'order_total', NOW(), NOW())
) VALUES ($1, $2, $3, $4::jsonb, 5000, 'order_total', NOW(), NOW())
RETURNING id`,
[orderId, skuNick, `集成测试SKU-${skuNick}`, JSON.stringify({ skuNick })],
[orderId, skuNick, skuNick, JSON.stringify({ skuNick })],
)
return { orderId, orderItemId: Number(item.rows[0]?.id) }
const orderItemId = Number(item.rows[0]?.id)
await query(
`UPDATE fulfillments
SET status = 'assigned',
channel = 'manual',
source = 'carrier_evidence',
updated_at = NOW()
WHERE order_item_id = $1`,
[orderItemId],
)
return { orderId, orderItemId }
}
function shanghaiDate(date = new Date()) {
apps/backend/src/services/finance/order-refund-service.ts
+12 -9
View File
@@ -81,7 +81,7 @@ export async function listOrderRefunds(orderId: number | string): Promise<OrderR
oi.sku_name,
COALESCE(NULLIF(u.display_name, ''), u.username, '') AS operator_name
FROM order_refunds r
LEFT JOIN order_items oi ON oi.id = r.order_item_id
LEFT JOIN v_order_item_current oi ON oi.id = r.order_item_id
LEFT JOIN admin_users u ON u.id = r.operator_admin_user_id
WHERE r.order_id = $1
ORDER BY r.refunded_at DESC, r.id DESC
@@ -144,7 +144,7 @@ export async function createManualRefund(
})
}
const activeItems = await client.query<{ total: number }>(
`SELECT COUNT(*)::int AS total FROM order_items
`SELECT COUNT(*)::int AS total FROM v_order_item_current
WHERE order_id = $1 AND fulfillment_status <> 'archived'`,
[orderId],
)
@@ -170,7 +170,7 @@ export async function createManualRefund(
FROM order_refunds r
WHERE r.order_item_id = oi.id AND r.status = 'confirmed'
), 0)::bigint AS refunded_fen
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.order_id = $1
ORDER BY oi.id
`,
@@ -286,7 +286,7 @@ export async function createManualRefund(
JOIN order_items ri ON ri.id = r.order_item_id
WHERE ri.order_id = $1 AND r.status = 'confirmed'
), 0)::bigint AS refunded_fen
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.order_id = $1
`,
[orderId],
@@ -407,7 +407,7 @@ export async function voidOrderRefund(
})
}
const activeItems = await client.query<{ total: number }>(
`SELECT COUNT(*)::int AS total FROM order_items
`SELECT COUNT(*)::int AS total FROM v_order_item_current
WHERE order_id = $1 AND fulfillment_status <> 'archived'`,
[row.order_id],
)
@@ -440,7 +440,7 @@ export async function voidOrderRefund(
JOIN order_items ri ON ri.id = r.order_item_id
WHERE ri.order_id = $1 AND r.status = 'confirmed'
), 0)::bigint AS refunded_fen
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.order_id = $1
`,
[row.order_id],
@@ -605,7 +605,7 @@ async function refreshActiveRefundFacts(itemIds: number[], basisDate: string): P
const ids = [...new Set(itemIds.filter((id) => Number.isInteger(id) && id > 0))]
if (ids.length === 0 || !basisDate) return
const active = await query<{ total: number }>(
`SELECT COUNT(*)::int AS total FROM order_items
`SELECT COUNT(*)::int AS total FROM v_order_item_current
WHERE id = ANY($1::bigint[]) AND fulfillment_status <> 'archived'`,
[ids],
)
@@ -684,7 +684,10 @@ export async function approveAndRecordKuaishouRefund(
}
const existingItem = await query<{ id: number; fulfillment_status: string }>(
`SELECT id, fulfillment_status FROM order_items WHERE id = $1 AND order_id = $2 LIMIT 1`,
`SELECT id, fulfillment_status
FROM v_order_item_current
WHERE id = $1 AND order_id = $2
LIMIT 1`,
[orderItemId, orderId],
)
if (!existingItem.rows[0]) {
@@ -897,7 +900,7 @@ async function getRefundableFen(orderId: number, orderItemId: number): Promise<n
FROM order_refunds r
WHERE r.order_item_id = oi.id AND r.status = 'confirmed'
), 0))::bigint AS remaining_fen
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.id = $1 AND oi.order_id = $2
LIMIT 1
`,
apps/backend/src/services/finance/refresh-facts.ts
+1 -1
View File
@@ -61,7 +61,7 @@ export async function orderItemBasisDates(itemIds: number[]): Promise<string[]>
SELECT DISTINCT
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
AS basis_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = ANY($1::bigint[])
`,
apps/backend/src/services/finance/sales-facts-channel-service.ts
+1 -1
View File
@@ -80,7 +80,7 @@ export async function rebuildChannelSalesDailyFacts(input: {
WHEN oi.fulfillment_channel = 'work_order' THEN 'sys_work_order_reward'
ELSE NULL
END AS channel_cost_code
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status IN ('assigned', 'pending')
AND o.pay_status IN ('paid', 'refunded')
apps/backend/src/services/fulfillment/affiliate-fallback-service.ts
+1 -1
View File
@@ -55,7 +55,7 @@ export async function runAffiliateFallback(
cfg.id AS config_id,
cfg.canonical_product_id AS config_product_id,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
apps/backend/src/services/fulfillment/carrier-evidence-service.ts
+1 -1
View File
@@ -44,7 +44,7 @@ export async function loadActiveCarrierEvidence(
AND ft.executor_key = 'manual_dispatch'
AND ft.task_status <> ALL($2::text[])
) AS has_manual_task
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.id = ANY($1::bigint[])
`,
[ids, INACTIVE_CARRIER_TASK_STATUSES],
apps/backend/src/services/fulfillment/carrier-reconciliation-service.ts
+6 -5
View File
@@ -105,7 +105,7 @@ async function loadCandidateItems(input: {
oi.order_id,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku,
capability.channels
FROM order_items oi
FROM v_order_item_current oi
JOIN LATERAL (
SELECT COALESCE(
jsonb_agg(
@@ -296,9 +296,10 @@ export async function reconcilePendingCarriers(
for (const [orderId, candidates] of byOrder) {
const order = await getOrderById(orderId)
if (!order) continue
const itemRows = await query<OrderItemRow>(`SELECT * FROM order_items WHERE order_id = $1`, [
orderId,
])
const itemRows = await query<OrderItemRow>(
`SELECT * FROM v_order_item_current WHERE order_id = $1`,
[orderId],
)
const itemById = new Map(itemRows.rows.map((item) => [Number(item.id), item]))
for (const candidate of candidates) {
@@ -424,7 +425,7 @@ export async function reconcilePendingCarriers(
const remainingResult = await query<{ count: number }>(
`
SELECT count(*)::int AS count
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.fulfillment_status IN ('pending', 'assigned')
AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
AND NOT EXISTS (
apps/backend/src/services/fulfillment/channel-assign-service.ts
+34 -17
View File
@@ -50,8 +50,8 @@ import {
/**
* 管道 A：渠道判定。
*
* 只写 `fulfillment_channel / fulfillment_status / fulfillment_source /
* fulfillment_decided_at`，绝不写商品、成本、事实，也绝不改旧 channel。
* 只写 `fulfillments.channel / status / source / decided_at`，
* 绝不写商品、成本、事实，也绝不改冻结的订单项旧列。
*
* 判定顺序：
* 1. 运营锁定 → 跳过
@@ -335,7 +335,7 @@ export async function assignPendingOrderItemChannels(
WHERE task.order_item_id = oi.id AND task.executor_key = 'manual_dispatch'
AND ${ACTIVE_CARRIER_TASK_SQL}
) AS has_manual_task
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'pending'
${options.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}
@@ -493,7 +493,9 @@ export async function assignOrderItemsAfterCarrierReady(input: {
}
const items = (
await query<OrderItemRow>(`SELECT * FROM order_items WHERE id = ANY($1::bigint[])`, [ids])
await query<OrderItemRow>(`SELECT * FROM v_order_item_current WHERE id = ANY($1::bigint[])`, [
ids,
])
).rows
const orderIds = [...new Set(items.map((item) => Number(item.order_id)))]
const orders = new Map(
@@ -712,9 +714,13 @@ async function syncAssignedChannelSnapshots(
|| jsonb_build_object('channel', $3)
),
updated_at = NOW()
WHERE id = ANY($1::bigint[])
AND fulfillment_status = 'assigned'
AND fulfillment_channel = $3`,
WHERE id IN (
SELECT id
FROM v_order_item_current
WHERE id = ANY($1::bigint[])
AND fulfillment_status = 'assigned'
AND fulfillment_channel = $3
)`,
[ids, JSON.stringify(decision), channel],
)
}
@@ -759,7 +765,7 @@ export async function switchChannelForOrderItems(input: {
const skuRows = await query<{ internal_sku: string }>(
`SELECT DISTINCT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.id = ANY($1::bigint[])`,
[ids],
)
@@ -783,7 +789,9 @@ export async function switchChannelForOrderItems(input: {
}
const items = (
await query<OrderItemRow>(`SELECT * FROM order_items WHERE id = ANY($1::bigint[])`, [ids])
await query<OrderItemRow>(`SELECT * FROM v_order_item_current WHERE id = ANY($1::bigint[])`, [
ids,
])
).rows
const orderIds = [...new Set(items.map((item) => Number(item.order_id)))]
const orders = new Map(
@@ -839,7 +847,10 @@ export async function switchChannelForOrderItems(input: {
const now = nowIso()
await withTransaction(async (client) => {
const locked = await client.query<OrderItemRow>(
`SELECT * FROM order_items WHERE id = $1 FOR UPDATE`,
`SELECT status AS fulfillment_status
FROM fulfillments
WHERE order_item_id = $1
FOR UPDATE`,
[item.id],
)
if (!locked.rows[0] || locked.rows[0].fulfillment_status === 'archived') {
@@ -983,7 +994,7 @@ export async function assignChannelToOrderItems(input: {
}
const rows = await query<{ internal_sku: string }>(
`SELECT resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.id = ANY($1::bigint[])`,
[ids],
)
@@ -1453,7 +1464,7 @@ export async function retryPendingCarrierGroup(input: {
oi.id AS order_item_id,
cfg.id AS config_id,
cfg.canonical_product_id
FROM order_items oi
FROM v_order_item_current oi
JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) =
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
@@ -1489,7 +1500,7 @@ export async function retryPendingCarrierGroup(input: {
`SELECT
count(*) FILTER (WHERE fulfillment_status = 'assigned')::int AS assigned,
count(*) FILTER (WHERE fulfillment_status = 'pending')::int AS remaining
FROM order_items
FROM v_order_item_current
WHERE id = ANY($1::bigint[])`,
[ids],
)
@@ -1611,7 +1622,7 @@ export async function deletePendingChannelGroup(input: {
const result = await withTransaction(async (client) => {
const itemParams: unknown[] = []
const itemWhere = ["oi.fulfillment_status = 'pending'"]
const itemWhere = ["fulfillment.status = 'pending'"]
if (scope.type === 'internal_sku') {
itemParams.push(scope.internalSku)
itemWhere.push(`${INTERNAL_SKU_SQL} = $${itemParams.length}`)
@@ -1637,10 +1648,11 @@ export async function deletePendingChannelGroup(input: {
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text
AS basis_date
FROM order_items oi
JOIN fulfillments fulfillment ON fulfillment.order_item_id = oi.id
JOIN orders o ON o.id = oi.order_id
WHERE ${itemWhere.join(' AND ')}
ORDER BY oi.id
FOR UPDATE OF oi
FOR UPDATE OF fulfillment
`,
itemParams,
)
@@ -1783,7 +1795,12 @@ export async function assignPendingChannelGroup(input: {
AND o.provider = $1
AND o.shop_id = $2
AND oi.sku_code = $3
AND oi.fulfillment_status = 'pending'
AND EXISTS (
SELECT 1
FROM fulfillments fulfillment
WHERE fulfillment.order_item_id = oi.id
AND fulfillment.status = 'pending'
)
${issueCategoryFilter}
RETURNING oi.id
`,
@@ -1943,7 +1960,7 @@ export async function assignChannelBySku(input: {
`
SELECT oi.id, ${INTERNAL_SKU_SQL} AS internal_sku
FROM orders o
JOIN order_items oi ON oi.order_id = o.id
JOIN v_order_item_current oi ON oi.order_id = o.id
WHERE o.id = oi.order_id
AND o.provider = $1
AND o.shop_id = $2
apps/backend/src/services/fulfillment/fulfillment-retry-guard.test.ts
+1 -1
View File
@@ -57,7 +57,7 @@ test('创建新履约尝试时自动退休旧当前尝试', () => {
assert.ok(taskRepo.includes('MAX(attempt_no)'))
assert.ok(taskRepo.includes('SET is_current = FALSE'))
assert.ok(taskRepo.includes('$26, TRUE'))
assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE'))
assert.ok(taskRepo.includes('SELECT id FROM fulfillments WHERE order_item_id = $1 FOR UPDATE'))
})
test('当前尝试必须是同单元最新任务，统一视图不能读到已退休尝试', () => {
apps/backend/src/services/fulfillment/fulfillment-routing.integration.test.ts
+62 -47
View File
@@ -206,7 +206,7 @@ test(
`UPDATE orders
SET created_at = '2019-01-01T00:00:00+08:00',
updated_at = '2019-01-01T00:00:00+08:00'
WHERE id = (SELECT order_id FROM order_items WHERE id = $1)`,
WHERE id = (SELECT order_id FROM v_order_item_current WHERE id = $1)`,
[cutoffOldItemId],
)
await setRoute({ sku: cutoffTestSku, channel: 'third_party_voucher' })
@@ -220,7 +220,7 @@ test(
assert.equal(cutoffAssignment.assigned, 1)
const cutoffStates = await query<{ id: number; fulfillment_status: string }>(
`SELECT id, fulfillment_status
FROM order_items
FROM v_order_item_current
WHERE id = ANY($1::bigint[])
ORDER BY id DESC`,
[[cutoffOldItemId, cutoffNewItemId]],
@@ -307,7 +307,7 @@ test(
fulfillment_config_id: number | null
}>(
`SELECT fulfillment_status, fulfillment_channel, fulfillment_config_id
FROM order_items WHERE id = $1`,
FROM v_order_item_current WHERE id = $1`,
[manualItemId],
)
assert.equal(manualItem.rows[0]?.fulfillment_status, 'assigned')
@@ -341,7 +341,7 @@ test(
}>(
`SELECT oi.fulfillment_status, oi.fulfillment_channel, oi.original_channel,
oi.canonical_product_id, product.canonical_sku, oi.item_snapshot_json
FROM order_items oi
FROM v_order_item_current oi
LEFT JOIN product_catalog product ON product.id = oi.canonical_product_id
WHERE oi.id = $1`,
[manualItemId],
@@ -370,7 +370,7 @@ test(
fulfillment_pending_reason: string | null
}>(
`SELECT fulfillment_status, fulfillment_channel, fulfillment_pending_reason
FROM order_items WHERE id = $1`,
FROM v_order_item_current WHERE id = $1`,
[missingAffiliateItemId],
)
assert.equal(pendingItem.rows[0]?.fulfillment_status, 'pending')
@@ -402,7 +402,7 @@ test(
canonical_product_id: number | null
}>(
`SELECT fulfillment_status, fulfillment_channel, fulfillment_config_id, canonical_product_id
FROM order_items WHERE id = $1`,
FROM v_order_item_current WHERE id = $1`,
[autoManualItemId],
)
assert.equal(autoManualItem.rows[0]?.fulfillment_status, 'assigned')
@@ -455,7 +455,7 @@ test(
work_rule_key: string | null
}>(
`SELECT oi.fulfillment_status, oi.fulfillment_channel, wpr.rule_key AS work_rule_key
FROM order_items oi
FROM v_order_item_current oi
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id
LEFT JOIN work_product_rules wpr ON wpr.id = wo.product_rule_id
WHERE oi.id = $1`,
@@ -470,11 +470,11 @@ test(
[retryWorkOrderItemId],
)
await query(
`UPDATE order_items
SET fulfillment_status = 'pending', fulfillment_channel = NULL,
fulfillment_source = 'auto_mapping', fulfillment_pending_reason = 'no_carrier',
fulfillment_pending_detail = $2::jsonb, updated_at = NOW()
WHERE id = $1`,
`UPDATE fulfillments
SET status = 'pending', channel = NULL,
source = 'auto_mapping', pending_reason = 'no_carrier',
pending_detail = $2::jsonb, updated_at = NOW()
WHERE order_item_id = $1`,
[
retryWorkOrderItemId,
JSON.stringify({ carrierIssue: 'carrier_task_closed', channel: 'work_order' }),
@@ -569,11 +569,11 @@ test(
shopName: '测试店铺 A',
})
await query(
`UPDATE order_items
SET fulfillment_status = 'assigned', fulfillment_channel = 'manual',
fulfillment_source = 'carrier_evidence', fulfillment_config_id = NULL,
`UPDATE fulfillments
SET status = 'assigned', channel = 'manual',
source = 'carrier_evidence', fulfillment_config_id = NULL,
updated_at = NOW()
WHERE id = $1`,
WHERE order_item_id = $1`,
[manualRemoveItemId],
)
await query(
@@ -596,7 +596,7 @@ test(
const removedManualItem = await query<{
fulfillment_status: string
fulfillment_channel: string | null
}>(`SELECT fulfillment_status, fulfillment_channel FROM order_items WHERE id = $1`, [
}>(`SELECT fulfillment_status, fulfillment_channel FROM v_order_item_current WHERE id = $1`, [
manualRemoveItemId,
])
assert.equal(removedManualItem.rows[0]?.fulfillment_status, 'pending')
@@ -617,11 +617,11 @@ test(
await setRoute({ sku: fallbackSku, channel: 'affiliate_dash' })
await setRoute({ sku: fallbackSku, channel: 'manual', role: 'fallback' })
const fallbackItem = await query<{ order_id: number }>(
`UPDATE order_items
SET fulfillment_status = 'assigned', fulfillment_channel = 'affiliate_dash',
fulfillment_source = 'operator_assigned', fulfillment_locked = FALSE,
`UPDATE fulfillments
SET status = 'assigned', channel = 'affiliate_dash',
source = 'operator_assigned', locked = FALSE,
updated_at = NOW()
WHERE id = $1
WHERE order_item_id = $1
RETURNING order_id`,
[fallbackItemId],
)
@@ -672,7 +672,7 @@ test(
assert.equal(
(
await query<{ fulfillment_channel: string }>(
`SELECT fulfillment_channel FROM order_items WHERE id = $1`,
`SELECT fulfillment_channel FROM v_order_item_current WHERE id = $1`,
[fallbackItemId],
)
).rows[0]?.fulfillment_channel,
@@ -694,7 +694,7 @@ test(
item_snapshot_json: Record<string, unknown>
}>(
`SELECT fulfillment_channel, fulfillment_config_id, item_snapshot_json
FROM order_items WHERE id = $1`,
FROM v_order_item_current WHERE id = $1`,
[fallbackItemId],
)
assert.equal(fallbackAssigned.rows[0]?.fulfillment_channel, 'manual')
@@ -715,7 +715,7 @@ test(
platform_order_id: string
}>(
`SELECT oi.order_id, o.provider, o.platform, o.shop_id, o.shop_name, o.platform_order_id
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = $1`,
[retryTaskItemId],
@@ -926,14 +926,22 @@ test(
configJson: { affiliateSku: retryPendingAffiliateSku },
})
await query(
`UPDATE order_items
SET fulfillment_status = 'assigned',
fulfillment_channel = 'affiliate_dash',
fulfillment_source = 'carrier_evidence',
`UPDATE fulfillments
SET status = 'assigned',
channel = 'affiliate_dash',
source = 'carrier_evidence',
updated_at = NOW()
WHERE id = $1`,
WHERE order_item_id = $1`,
[retryPendingItemId],
)
const existingRetryTasks = await query<{ id: number; task_status: string }>(
`SELECT id, task_status
FROM fulfillment_tasks
WHERE order_item_id = $1
ORDER BY id`,
[retryPendingItemId],
)
assert.deepEqual(existingRetryTasks.rows, [], `测试订单项已有任务：${retryPendingItemId}`)
await query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
@@ -943,16 +951,23 @@ test(
SELECT oi.order_id, oi.id, 1, $2, o.provider, o.platform,
o.shop_id, o.shop_name, o.platform_order_id, profile.id, 'affiliate_dash',
'retry_pending', NOW(), NOW()
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
JOIN LATERAL (
SELECT id FROM fulfillment_profiles
WHERE profile_key = 'affiliate_dash' AND executor_key = 'affiliate_dash'
ORDER BY id LIMIT 1
) profile ON TRUE
JOIN fulfillment_profiles profile
ON profile.id = $3
AND profile.executor_key = 'affiliate_dash'
WHERE oi.id = $1`,
[retryPendingItemId, `IT-RETRY-PENDING-${suffix}`],
[retryPendingItemId, `IT-RETRY-PENDING-${suffix}`, fallbackProfileId],
)
const retryPendingTask = await query<{ task_status: string }>(
`SELECT task_status
FROM fulfillment_tasks
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'
ORDER BY id DESC
LIMIT 1`,
[retryPendingItemId],
)
assert.equal(retryPendingTask.rows[0]?.task_status, 'retry_pending')
const blockedRetry = await reconcilePendingCarriers({
orderItemIds: [retryPendingItemId],
includeAssignedMissingCarrier: true,
@@ -972,7 +987,7 @@ test(
oi.fulfillment_pending_reason,
oi.fulfillment_pending_detail,
(SELECT count(*)::int FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) AS task_count
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.id = $1`,
[retryPendingItemId],
)
@@ -985,7 +1000,7 @@ test(
)
assert.equal(retryPendingState.rows[0]?.task_count, 1)
await query(
`UPDATE order_items SET fulfillment_locked = TRUE, updated_at = NOW() WHERE id = $1`,
`UPDATE fulfillments SET locked = TRUE, updated_at = NOW() WHERE order_item_id = $1`,
[retryPendingItemId],
)
const unlockedPendingGroup = await unlockPendingChannelGroup({
@@ -993,7 +1008,7 @@ test(
})
assert.equal(unlockedPendingGroup.updated, 1)
const unlockedPendingItem = await query<{ fulfillment_locked: boolean }>(
`SELECT fulfillment_locked FROM order_items WHERE id = $1`,
`SELECT fulfillment_locked FROM v_order_item_current WHERE id = $1`,
[retryPendingItemId],
)
assert.equal(unlockedPendingItem.rows[0]?.fulfillment_locked, false)
@@ -1007,14 +1022,14 @@ test(
[retryPendingItemId],
)
await query(
`UPDATE order_items
SET fulfillment_status = 'pending',
fulfillment_channel = NULL,
fulfillment_source = 'auto_mapping',
fulfillment_pending_reason = NULL,
fulfillment_pending_detail = NULL,
`UPDATE fulfillments
SET status = 'pending',
channel = NULL,
source = 'auto_mapping',
pending_reason = NULL,
pending_detail = NULL,
updated_at = NOW()
WHERE id = $1`,
WHERE order_item_id = $1`,
[retryPendingItemId],
)
await assert.rejects(
@@ -1035,7 +1050,7 @@ test(
oi.fulfillment_status,
oi.fulfillment_pending_detail,
(SELECT count(*)::int FROM fulfillment_tasks ft WHERE ft.order_item_id = oi.id) AS task_count
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.id = $1`,
[retryPendingItemId],
)
apps/backend/src/services/fulfillment/fulfillment-shadow-model.test.ts
-73
View File
@@ -1,73 +0,0 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const migration = readFileSync(
new URL('../../db/migrations/141_fulfillment_shadow_model.sql', import.meta.url),
'utf8',
)
const auditScript = readFileSync(
new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url),
'utf8',
)
const configSyncMigration = readFileSync(
new URL('../../db/migrations/144_fulfillment_shadow_configuration_sync.sql', import.meta.url),
'utf8',
)
const routeCutoverMigration = readFileSync(
new URL('../../db/migrations/161_move_shadow_sync_to_sku_routes.sql', import.meta.url),
'utf8',
)
test('正式履约影子模型包含履约、尝试和异常三张表', () => {
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillments'))
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_attempts'))
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_issues'))
assert.ok(migration.includes('uq_fulfillment_attempts_current_unit'))
assert.ok(migration.includes('UNIQUE (fulfillment_id, issue_code)'))
})
test('影子模型从现有真源回填并持续双写', () => {
assert.ok(migration.includes('SELECT sync_fulfillment_shadow(state.order_item_id)'))
assert.ok(migration.includes('trg_order_items_fulfillment_shadow'))
assert.ok(migration.includes('trg_fulfillment_tasks_shadow'))
assert.ok(migration.includes('trg_work_orders_fulfillment_shadow'))
assert.ok(migration.includes('assign_fulfillment_event_shadow'))
})
test('人工配置和统一渠道规则变化会刷新履约影子', () => {
assert.ok(configSyncMigration.includes('trg_manual_fulfillment_config_shadow'))
assert.ok(configSyncMigration.includes('sync_fulfillment_shadow(item.id)'))
assert.ok(routeCutoverMigration.includes('trg_sku_route_fulfillment_shadow'))
assert.ok(routeCutoverMigration.includes('sync_fulfillment_shadow_for_sku_route'))
assert.ok(
routeCutoverMigration.includes('DROP TRIGGER IF EXISTS trg_internal_sku_channel_shadow'),
)
})
test('人工履约配置不能关联其他来源商品', () => {
assert.ok(configSyncMigration.includes('trg_manual_config_product_source'))
assert.ok(configSyncMigration.includes("product.canonical_sku LIKE 'man:%'"))
assert.ok(configSyncMigration.includes("ERRCODE = '23514'"))
assert.ok(configSyncMigration.includes('BEFORE INSERT OR UPDATE ON manual_fulfillment_configs'))
assert.ok(configSyncMigration.includes('config.enabled = TRUE'))
assert.ok(configSyncMigration.includes("SET status = 'active'"))
})
test('明确渠道即视为已路由，不用渠道配置缺失覆盖已选渠道', () => {
assert.ok(migration.includes('v_fulfillment_item_state_raw'))
assert.ok(migration.includes("WHEN raw.intended_channel IS NOT NULL THEN 'configured'"))
assert.ok(
migration.includes(
"WHEN raw.route_status = 'missing_channel'\n AND raw.intended_channel IS NOT NULL\n THEN ''",
),
)
})
test('审计脚本核对影子表、执行尝试、开放异常和事件关联', () => {
assert.ok(auditScript.includes('影子履约缺失'))
assert.ok(auditScript.includes('影子履约状态差异'))
assert.ok(auditScript.includes('影子执行尝试缺失'))
assert.ok(auditScript.includes('影子开放异常缺失'))
assert.ok(auditScript.includes('履约事件未关联影子记录'))
})
apps/backend/src/services/fulfillment/fulfillment-state-service.integration.test.ts
+7 -7
View File
@@ -116,10 +116,10 @@ test(
assert.equal((await listFulfillmentStateEvents(orderItemId)).length, 1)
await query(
`UPDATE order_items
SET fulfillment_pending_detail = jsonb_build_object('legacyWrite', TRUE),
`UPDATE fulfillments
SET pending_detail = jsonb_build_object('legacyWrite', TRUE),
updated_at = NOW()
WHERE id = $1`,
WHERE order_item_id = $1`,
[orderItemId],
)
events = await listFulfillmentStateEvents(orderItemId)
@@ -128,10 +128,10 @@ test(
assert.equal(events[0]?.actor_type, 'system')
await query(
`UPDATE order_items
SET fulfillment_pending_detail = jsonb_build_object('legacyWrite', TRUE),
`UPDATE fulfillments
SET pending_detail = jsonb_build_object('legacyWrite', TRUE),
updated_at = NOW()
WHERE id = $1`,
WHERE order_item_id = $1`,
[orderItemId],
)
events = await listFulfillmentStateEvents(orderItemId)
@@ -193,7 +193,7 @@ test(
attributes_json, created_at, updated_at
) VALUES ($1, $2, '测试', '件', 'active', '{}'::jsonb, NOW(), NOW())
RETURNING id`,
[`test:ff-state-${suffix}`, `状态事件商品-${suffix}`],
[`man:ff-state-${suffix}`, `状态事件商品-${suffix}`],
)
productId = Number(product.rows[0]!.id)
const config = await query<{ id: number }>(
apps/backend/src/services/fulfillment/fulfillment-state-service.test.ts
+17 -4
View File
@@ -7,6 +7,7 @@ function read(relativePath: string) {
}
const migration = read('../../db/migrations/140_fulfillment_state_events.sql')
const cutoverMigration = read('../../db/migrations/163_cutover_fulfillment_truth.sql')
const stateService = read('./fulfillment-state-service.ts')
const channelAssign = read('./channel-assign-service.ts')
const carrierReconciliation = read('./carrier-reconciliation-service.ts')
@@ -32,8 +33,10 @@ test('履约状态事件表记录前后状态并禁止应用修改', () => {
assert.ok(migration.includes('fulfillment_state_events is append-only'))
})
test('状态写入口串行锁订单项，并在写入前校验期望状态', () => {
assert.ok(stateService.includes('SELECT * FROM order_items WHERE id = $1 FOR UPDATE'))
test('状态写入口串行锁正式履约记录，并在写入前校验期望状态', () => {
assert.ok(stateService.includes('FROM fulfillments'))
assert.ok(stateService.includes('FOR UPDATE'))
assert.ok(!stateService.includes('SELECT * FROM order_items WHERE id = $1 FOR UPDATE'))
assert.ok(stateService.includes('expected.status'))
assert.ok(stateService.includes('expected.channel'))
assert.ok(stateService.includes('sourceNotIn'))
@@ -69,8 +72,18 @@ test('履约配置和标准商品关联统一通过状态服务的映射接口
assert.ok(stateService.includes('onlyWhenCanonicalProductNull'))
})
test('任务创建先锁订单项并复用已有活跃尝试', () => {
assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE'))
test('任务创建先锁正式履约记录并复用已有活跃尝试', () => {
assert.ok(taskRepo.includes('SELECT id FROM fulfillments WHERE order_item_id = $1 FOR UPDATE'))
assert.ok(taskRepo.includes('task_status NOT IN ('))
assert.ok(taskRepo.includes('ORDER BY is_current DESC, id DESC'))
})
test('订单项状态列冻结，读模型直接投影正式履约表', () => {
assert.ok(
cutoverMigration.includes('RENAME COLUMN fulfillment_status TO legacy_fulfillment_status'),
)
assert.ok(cutoverMigration.includes('CREATE VIEW v_order_item_current'))
assert.ok(cutoverMigration.includes('FROM fulfillments fulfillment'))
assert.ok(cutoverMigration.includes('DROP FUNCTION IF EXISTS sync_fulfillment_shadow(BIGINT)'))
assert.ok(cutoverMigration.includes('CREATE TRIGGER trg_fulfillments_state_event'))
})
apps/backend/src/services/fulfillment/fulfillment-state-service.ts
+140 -77
View File
@@ -97,7 +97,6 @@ type StateColumn = {
column: string
value: unknown
cast?: 'jsonb'
mergeSnapshot?: boolean
coalesceCurrent?: boolean
preserveCarrierIssue?: boolean
}
@@ -119,46 +118,46 @@ function buildPatchColumns(patch: FulfillmentStatePatch): StateColumn[] {
if (!['pending', 'assigned', 'archived'].includes(patch.status)) {
throw new TypeError(`invalid fulfillment status: ${patch.status}`)
}
columns.push({ column: 'fulfillment_status', value: patch.status })
columns.push({ column: 'status', value: patch.status })
}
if (patch.channel !== undefined) {
if (patch.channel !== null && !isFulfillmentChannel(patch.channel)) {
throw new TypeError(`invalid fulfillment channel: ${patch.channel}`)
}
columns.push({ column: 'fulfillment_channel', value: patch.channel })
columns.push({ column: 'channel', value: patch.channel })
}
if (patch.source !== undefined) {
columns.push({ column: 'fulfillment_source', value: patch.source })
columns.push({ column: 'source', value: patch.source })
}
if (patch.locked !== undefined) {
columns.push({ column: 'fulfillment_locked', value: Boolean(patch.locked) })
columns.push({ column: 'locked', value: Boolean(patch.locked) })
}
if (patch.pendingReason !== undefined) {
columns.push({ column: 'fulfillment_pending_reason', value: patch.pendingReason || null })
columns.push({ column: 'pending_reason', value: patch.pendingReason || null })
}
if (patch.pendingDetail !== undefined) {
columns.push({
column: 'fulfillment_pending_detail',
column: 'pending_detail',
value: patch.pendingDetail,
cast: 'jsonb',
})
}
if (patch.pendingDetailUnlessCarrierIssue !== undefined) {
columns.push({
column: 'fulfillment_pending_detail',
column: 'pending_detail',
value: patch.pendingDetailUnlessCarrierIssue,
cast: 'jsonb',
preserveCarrierIssue: true,
})
}
if (patch.reason !== undefined) {
columns.push({ column: 'fulfillment_reason', value: patch.reason || null })
columns.push({ column: 'reason', value: patch.reason || null })
}
if (patch.decidedBy !== undefined) {
columns.push({ column: 'fulfillment_decided_by', value: patch.decidedBy })
columns.push({ column: 'decided_by', value: patch.decidedBy })
}
if (patch.decidedAt !== undefined) {
columns.push({ column: 'fulfillment_decided_at', value: patch.decidedAt })
columns.push({ column: 'decided_at', value: patch.decidedAt })
}
if (patch.originalChannelIfNull !== undefined) {
if (!isFulfillmentChannel(patch.originalChannelIfNull)) {
@@ -197,26 +196,49 @@ function buildPatchColumns(patch: FulfillmentStatePatch): StateColumn[] {
})
}
if (patch.itemSnapshotJson !== undefined) {
columns.push({
column: 'item_snapshot_json',
value: patch.itemSnapshotJson,
cast: 'jsonb',
})
// 原订单项快照属于订单事实，不进入履约状态表。
}
if (patch.itemSnapshotPatch !== undefined) {
columns.push({
column: 'item_snapshot_json',
value: patch.itemSnapshotPatch,
cast: 'jsonb',
mergeSnapshot: true,
})
// 同上，在状态更新后单独合并。
}
if (columns.length === 0) {
if (
columns.length === 0 &&
patch.itemSnapshotJson === undefined &&
patch.itemSnapshotPatch === undefined
) {
throw new TypeError('fulfillment state patch must change at least one column')
}
return columns
}
async function applyOrderItemSnapshotPatch(
executor: FulfillmentStateExecutor,
orderItemId: number,
patch: FulfillmentStatePatch,
) {
if (patch.itemSnapshotJson === undefined && patch.itemSnapshotPatch === undefined) {
return
}
if (patch.itemSnapshotJson !== undefined) {
await executor.query(
`UPDATE order_items
SET item_snapshot_json = $2::jsonb,
updated_at = $3::timestamptz
WHERE id = $1`,
[orderItemId, patch.itemSnapshotJson, new Date().toISOString()],
)
return
}
await executor.query(
`UPDATE order_items
SET item_snapshot_json =
COALESCE(item_snapshot_json, '{}'::jsonb) || $2::jsonb,
updated_at = $3::timestamptz
WHERE id = $1`,
[orderItemId, patch.itemSnapshotPatch, new Date().toISOString()],
)
}
async function setEventContext(
executor: FulfillmentStateExecutor,
transition: Pick<FulfillmentStateTransition, 'eventType'> & {
@@ -261,71 +283,84 @@ export async function transitionFulfillmentStateWithExecutor(
const columns = buildPatchColumns(input.patch)
await setEventContext(executor, input)
try {
const lockResult = await executor.query<OrderItemRow>(
`SELECT * FROM order_items WHERE id = $1 FOR UPDATE`,
const lockResult = await executor.query<{
id: number
order_item_id: number
status: string
channel: string | null
source: string
locked: boolean
}>(
`SELECT id, order_item_id, status, channel, source, locked
FROM fulfillments
WHERE order_item_id = $1
FOR UPDATE`,
[orderItemId],
)
const current = lockResult.rows[0]
if (!current) return null
if (
input.expected?.status !== undefined &&
current.fulfillment_status !== input.expected.status
) {
if (input.expected?.status !== undefined && current.status !== input.expected.status) {
return null
}
if (
input.expected?.channel !== undefined &&
(current.fulfillment_channel ?? null) !== input.expected.channel
(current.channel ?? null) !== input.expected.channel
) {
return null
}
if (
input.expected?.source !== undefined &&
(current.fulfillment_source || '') !== (input.expected.source || '')
(current.source || '') !== (input.expected.source || '')
) {
return null
}
if (
input.expected?.sourceNotIn &&
input.expected.sourceNotIn.includes(String(current.fulfillment_source || ''))
input.expected.sourceNotIn.includes(String(current.source || ''))
) {
return null
}
if (
input.expected?.locked !== undefined &&
Boolean(current.fulfillment_locked) !== Boolean(input.expected.locked)
Boolean(current.locked) !== Boolean(input.expected.locked)
) {
return null
}
const params: unknown[] = []
const assignments = columns.map((entry) => {
params.push(entry.value)
let expression = `$${params.length}`
if (entry.cast) expression += `::${entry.cast}`
if (entry.mergeSnapshot) {
expression = `COALESCE(item_snapshot_json, '{}'::jsonb) || ${expression}`
} else if (entry.preserveCarrierIssue) {
expression =
`CASE WHEN COALESCE(fulfillment_pending_detail->>'carrierIssue', '') ` +
`NOT IN ('', 'carrier_not_attempted') ` +
`THEN fulfillment_pending_detail ELSE ${expression} END`
} else if (entry.coalesceCurrent) {
expression = `COALESCE(${entry.column}, ${expression})`
}
return `${entry.column} = ${expression}`
})
params.push(new Date().toISOString())
assignments.push(`updated_at = $${params.length}::timestamptz`)
params.push(orderItemId)
if (columns.length > 0) {
const params: unknown[] = []
const assignments = columns.map((entry) => {
params.push(entry.value)
let expression = `$${params.length}`
if (entry.cast) expression += `::${entry.cast}`
if (entry.preserveCarrierIssue) {
expression =
`CASE WHEN COALESCE(pending_detail->>'carrierIssue', '') ` +
`NOT IN ('', 'carrier_not_attempted') ` +
`THEN pending_detail ELSE ${expression} END`
} else if (entry.coalesceCurrent) {
expression = `COALESCE(${entry.column}, ${expression})`
}
return `${entry.column} = ${expression}`
})
params.push(new Date().toISOString())
assignments.push(`updated_at = $${params.length}::timestamptz`)
params.push(current.id)
await executor.query(
`UPDATE fulfillments
SET ${assignments.join(', ')}
WHERE id = $${params.length}`,
params,
)
}
await applyOrderItemSnapshotPatch(executor, orderItemId, input.patch)
await executor.query(`SELECT refresh_fulfillment_derived($1)`, [orderItemId])
const updated = await executor.query<OrderItemRow>(
`UPDATE order_items
SET ${assignments.join(', ')}
WHERE id = $${params.length}
RETURNING *`,
params,
`SELECT * FROM v_order_item_current WHERE id = $1`,
[orderItemId],
)
return updated.rows[0] || null
} finally {
@@ -409,21 +444,21 @@ export async function assignFulfillmentChannelsWithExecutor(
...(input.reason ? { reason: input.reason } : {}),
})
try {
const result = await executor.query<OrderItemRow>(
`UPDATE order_items item
SET fulfillment_channel = input.channel,
fulfillment_status = 'assigned',
fulfillment_source = input.source,
fulfillment_decided_at = NOW(),
fulfillment_pending_reason = NULL,
fulfillment_pending_detail = NULL,
const result = await executor.query<{ order_item_id: number }>(
`UPDATE fulfillments fulfillment
SET channel = input.channel,
status = 'assigned',
source = input.source,
decided_at = NOW(),
pending_reason = NULL,
pending_detail = NULL,
updated_at = NOW()
FROM unnest($1::bigint[], $2::text[], $3::text[])
AS input(order_item_id, channel, source)
WHERE item.id = input.order_item_id
AND item.fulfillment_status = 'pending'
AND (item.fulfillment_locked = FALSE OR $4::boolean)
RETURNING item.*`,
WHERE fulfillment.order_item_id = input.order_item_id
AND fulfillment.status = 'pending'
AND (fulfillment.locked = FALSE OR $4::boolean)
RETURNING fulfillment.order_item_id`,
[
assignments.map((row) => row.orderItemId),
assignments.map((row) => row.channel),
@@ -431,7 +466,21 @@ export async function assignFulfillmentChannelsWithExecutor(
input.includeLocked === true,
],
)
return result.rows
const updatedIds = result.rows.map((row) => Number(row.order_item_id))
if (updatedIds.length === 0) return []
await executor.query(
`SELECT refresh_fulfillment_derived(order_item_id)
FROM fulfillments
WHERE order_item_id = ANY($1::bigint[])`,
[updatedIds],
)
const updated = await executor.query<OrderItemRow>(
`SELECT * FROM v_order_item_current
WHERE id = ANY($1::bigint[])
ORDER BY id`,
[updatedIds],
)
return updated.rows
} finally {
await clearEventContext(executor)
}
@@ -487,14 +536,14 @@ export async function attachFulfillmentConfigurationsWithExecutor(
mappings.map((row) => row.requireExpectedProduct),
mappings.map((row) => row.expectedCanonicalProductId),
]
const where = ['item.id = input.order_item_id']
const where = ['item.order_item_id = input.order_item_id']
if (input.expected?.status !== undefined) {
params.push(input.expected.status)
where.push(`item.fulfillment_status IS NOT DISTINCT FROM $${params.length}`)
where.push(`item.status IS NOT DISTINCT FROM $${params.length}`)
}
if (input.expected?.channel !== undefined) {
params.push(input.expected.channel)
where.push(`item.fulfillment_channel IS NOT DISTINCT FROM $${params.length}`)
where.push(`item.channel IS NOT DISTINCT FROM $${params.length}`)
}
if (input.onlyWhenConfigNull) {
where.push('item.fulfillment_config_id IS NULL')
@@ -514,8 +563,8 @@ export async function attachFulfillmentConfigurationsWithExecutor(
params.push(new Date().toISOString())
const updatedAtParam = params.length
const result = await executor.query<OrderItemRow>(
`UPDATE order_items item
const updatedIds = await executor.query<{ order_item_id: number }>(
`UPDATE fulfillments item
SET fulfillment_config_id = CASE
WHEN input.set_config THEN input.config_id
ELSE item.fulfillment_config_id
@@ -545,9 +594,23 @@ export async function attachFulfillmentConfigurationsWithExecutor(
expected_canonical_product_id
)
WHERE ${where.join(' AND ')}
RETURNING item.*`,
RETURNING item.order_item_id`,
params,
)
const ids = updatedIds.rows.map((row) => Number(row.order_item_id))
if (ids.length === 0) return []
await executor.query(
`SELECT refresh_fulfillment_derived(order_item_id)
FROM fulfillments
WHERE order_item_id = ANY($1::bigint[])`,
[ids],
)
const result = await executor.query<OrderItemRow>(
`SELECT * FROM v_order_item_current
WHERE id = ANY($1::bigint[])
ORDER BY id`,
[ids],
)
return result.rows
}
apps/backend/src/services/fulfillment/fulfillment-truth-model.test.ts
+89
View File
@@ -0,0 +1,89 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const migration = readFileSync(
new URL('../../db/migrations/141_fulfillment_shadow_model.sql', import.meta.url),
'utf8',
)
const auditScript = readFileSync(
new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url),
'utf8',
)
const configSyncMigration = readFileSync(
new URL('../../db/migrations/144_fulfillment_shadow_configuration_sync.sql', import.meta.url),
'utf8',
)
const routeCutoverMigration = readFileSync(
new URL('../../db/migrations/161_move_shadow_sync_to_sku_routes.sql', import.meta.url),
'utf8',
)
const truthCutoverMigration = readFileSync(
new URL('../../db/migrations/163_cutover_fulfillment_truth.sql', import.meta.url),
'utf8',
)
const insertSkuMigration = readFileSync(
new URL('../../db/migrations/166_fulfillment_insert_resolved_sku.sql', import.meta.url),
'utf8',
)
test('正式履约模型包含履约、尝试和异常三张表', () => {
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillments'))
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_attempts'))
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_issues'))
assert.ok(migration.includes('uq_fulfillment_attempts_current_unit'))
assert.ok(migration.includes('UNIQUE (fulfillment_id, issue_code)'))
})
test('迁移 163 删除影子双写并直接维护正式履约真源', () => {
assert.ok(
truthCutoverMigration.includes('DROP TRIGGER IF EXISTS trg_order_items_fulfillment_shadow'),
)
assert.ok(
truthCutoverMigration.includes('DROP FUNCTION IF EXISTS sync_fulfillment_shadow(BIGINT)'),
)
assert.ok(truthCutoverMigration.includes('CREATE TRIGGER trg_fulfillments_state_event'))
assert.ok(truthCutoverMigration.includes('CREATE TRIGGER trg_fulfillment_tasks_attempt'))
assert.ok(truthCutoverMigration.includes('CREATE TRIGGER trg_work_orders_attempt'))
assert.ok(truthCutoverMigration.includes('CREATE VIEW v_order_item_current'))
})
test('人工配置和路由变化直接刷新正式履约投影', () => {
assert.ok(truthCutoverMigration.includes('trg_manual_config_refresh_fulfillments'))
assert.ok(truthCutoverMigration.includes('trg_sku_route_refresh_fulfillments'))
assert.ok(truthCutoverMigration.includes('refresh_fulfillment_derived'))
assert.ok(
routeCutoverMigration.includes('DROP TRIGGER IF EXISTS trg_internal_sku_channel_shadow'),
)
})
test('人工履约配置不能关联其他来源商品', () => {
assert.ok(configSyncMigration.includes('trg_manual_config_product_source'))
assert.ok(configSyncMigration.includes("product.canonical_sku LIKE 'man:%'"))
assert.ok(configSyncMigration.includes("ERRCODE = '23514'"))
assert.ok(configSyncMigration.includes('BEFORE INSERT OR UPDATE ON manual_fulfillment_configs'))
assert.ok(configSyncMigration.includes('config.enabled = TRUE'))
assert.ok(configSyncMigration.includes("SET status = 'active'"))
})
test('履约读模型直接投影正式表，订单项旧列冻结', () => {
assert.ok(
truthCutoverMigration.includes('RENAME COLUMN fulfillment_status TO legacy_fulfillment_status'),
)
assert.ok(
truthCutoverMigration.includes(
'RENAME COLUMN canonical_product_id TO legacy_canonical_product_id',
),
)
assert.ok(truthCutoverMigration.includes('FROM fulfillments fulfillment'))
assert.ok(truthCutoverMigration.includes('FROM fulfillment_attempts attempt'))
assert.ok(insertSkuMigration.includes('COALESCE(resolve_internal_sku'))
})
test('审计脚本核对正式履约、执行尝试、开放异常和事件关联', () => {
assert.ok(auditScript.includes('正式履约缺失'))
assert.ok(auditScript.includes('正式履约状态差异'))
assert.ok(auditScript.includes('执行尝试缺失'))
assert.ok(auditScript.includes('开放异常缺失'))
assert.ok(auditScript.includes('履约事件未关联正式记录'))
})
apps/backend/src/services/fulfillment/internal-sku-routing.integration.test.ts
+2 -2
View File
@@ -30,7 +30,7 @@ test(
WHERE trigger_name IN (
'trg_sync_sku_channel_routes',
'trg_internal_sku_channel_shadow',
'trg_sku_route_fulfillment_shadow'
'trg_sku_route_refresh_fulfillments'
)
ORDER BY event_object_table, trigger_name
`,
@@ -38,7 +38,7 @@ test(
assert.deepEqual(triggers.rows, [
{
event_object_table: 'sku_channel_routes',
trigger_name: 'trg_sku_route_fulfillment_shadow',
trigger_name: 'trg_sku_route_refresh_fulfillments',
},
])
apps/backend/src/services/fulfillment/pending-channel-repair-service.ts
+2 -2
View File
@@ -35,7 +35,7 @@ export async function retryPendingWorkOrderRouting(
if (internalSku) params.push(internalSku)
const pending = await query<OrderItemRow>(
`SELECT oi.*
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.fulfillment_status = 'pending'
AND EXISTS (
SELECT 1
@@ -98,7 +98,7 @@ export async function retryPendingWorkOrderRouting(
})
const remaining = await query<{ count: number }>(
`SELECT COUNT(*)::int AS count
FROM order_items
FROM v_order_item_current
WHERE id = ANY($1::bigint[]) AND fulfillment_status = 'pending'`,
[ids],
)
apps/backend/src/services/fulfillment/voucher-consume-queue-service.ts
+2 -2
View File
@@ -212,7 +212,7 @@ export async function scheduleVoucherConsumeForOrderItems(
o.provider,
o.platform,
o.platform_order_id
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = ANY($1::bigint[])
AND oi.fulfillment_status = 'assigned'
@@ -275,7 +275,7 @@ export async function scheduleVoucherConsumeFallbackForOrderItems(
o.provider,
o.platform,
o.platform_order_id
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
WHERE oi.id = ANY($1::bigint[])
AND oi.fulfillment_status = 'pending'
apps/backend/src/services/platforms/ninetyone/order-service.ts
+1 -1
View File
@@ -317,7 +317,7 @@ export async function listOpen91Orders({
'quantity', oi.quantity,
'snapshot', oi.item_snapshot_json
) ORDER BY oi.id ASC) AS items
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.order_id = o.id
) item_summary ON TRUE
LEFT JOIN LATERAL (
apps/backend/src/services/scheduler/finance-pipeline-jobs.ts
+9 -9
View File
@@ -131,7 +131,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
SELECT oi.id,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key,
cfg.id AS config_id
FROM order_items oi
FROM v_order_item_current oi
JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
@@ -169,7 +169,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
`
SELECT oi.id,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
AND oi.fulfillment_config_id IS NULL
@@ -210,7 +210,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
`
SELECT oi.id,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
AND oi.fulfillment_config_id IS NOT NULL
@@ -257,7 +257,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
`
SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku
FROM order_items oi
FROM v_order_item_current oi
LEFT JOIN work_orders wo ON wo.order_item_id = oi.id AND wo.status <> 'cancelled'
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
@@ -316,7 +316,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
`
SELECT oi.id, oi.fulfillment_channel AS channel, oi.sku_code,
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS internal_sku
FROM order_items oi
FROM v_order_item_current oi
WHERE oi.fulfillment_status = 'assigned'
AND (
EXISTS (SELECT 1 FROM work_orders wo WHERE wo.order_item_id = oi.id AND wo.status <> 'cancelled')
@@ -407,7 +407,7 @@ async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promis
SELECT DISTINCT ON (oi.id)
oi.id AS order_item_id,
cfg.id AS config_id
FROM order_items oi
FROM v_order_item_current oi
JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
@@ -456,7 +456,7 @@ async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promis
SELECT DISTINCT i.order_item_id
FROM cost_exception_items i
JOIN cost_exceptions ce ON ce.id = i.exception_id
JOIN order_items oi ON oi.id = i.order_item_id
JOIN v_order_item_current oi ON oi.id = i.order_item_id
WHERE ce.reason = 'carrier_missing'
AND ce.status <> 'resolved'
AND (
@@ -479,7 +479,7 @@ async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promis
SELECT DISTINCT i.order_item_id
FROM cost_exception_items i
JOIN cost_exceptions ce ON ce.id = i.exception_id
JOIN order_items oi ON oi.id = i.order_item_id
JOIN v_order_item_current oi ON oi.id = i.order_item_id
WHERE ce.reason = 'manual_config_missing'
AND ce.status <> 'resolved'
AND oi.fulfillment_config_id IS NOT NULL
@@ -493,7 +493,7 @@ async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promis
SELECT DISTINCT i.order_item_id
FROM cost_exception_items i
JOIN cost_exceptions ce ON ce.id = i.exception_id
JOIN order_items oi ON oi.id = i.order_item_id
JOIN v_order_item_current oi ON oi.id = i.order_item_id
WHERE ce.reason = 'manual_cost_missing'
AND ce.status <> 'resolved'
AND EXISTS (
apps/backend/src/services/worker-platform/admin-kuaishou-match-service.ts
+1 -1
View File
@@ -210,7 +210,7 @@ export async function listAdminKuaishouMatchSources(payload: JsonObject = {}) {
oi.item_snapshot_json,
oi.created_at,
oi.updated_at
FROM order_items oi
FROM v_order_item_current oi
JOIN orders o ON o.id = oi.order_id
ORDER BY oi.id DESC
LIMIT $1