feat(finance): 商品渠道能力、自动发货转人工与成本异常收口

- 商品可同时支持自动发货/人工,新增渠道能力与成本冲销字段
- 卡券缺商品时按内部中文 SKU 自动建 tpv: 商品并补渠道规则
- 自动发货失败转人工并冲销原渠道成本
- 配置成本后按最早缺失订单生效并立即补算、关闭异常
- tpv: 商品来源计入第三方卡券
This commit is contained in:
yml2213
2026-11-02 13:17:15 +08:00
parent 86b5347e40
commit 3c4c81003f
16 changed files with 532 additions and 34 deletions
@@ -0,0 +1,100 @@
-- 115_product_channel_capabilities.sql
--
-- 渠道能力与履约兜底:
-- 1. 商品可同时具备多个渠道能力(自动发货 + 人工兜底);
-- 2. 订单项记录原始渠道与切换原因;
-- 3. 成本快照支持冲销(reversed_at),切换渠道时不双算;
-- 4. fulfillment_source 增加 affiliate_fallback。
CREATE TABLE IF NOT EXISTS product_channel_capabilities (
product_id BIGINT NOT NULL REFERENCES product_catalog(id) ON DELETE CASCADE,
channel TEXT NOT NULL,
priority INTEGER NOT NULL DEFAULT 100,
is_fallback BOOLEAN NOT NULL DEFAULT FALSE,
enabled BOOLEAN NOT NULL DEFAULT TRUE,
source TEXT NOT NULL DEFAULT 'auto',
created_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
updated_at TIMESTAMPTZ NOT NULL DEFAULT NOW(),
PRIMARY KEY (product_id, channel)
);
COMMENT ON TABLE product_channel_capabilities IS
'商品渠道能力:一个商品可同时支持自动发货、人工等;priority 越小越优先';
-- 自动发货商品:默认自动优先,人工兜底
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback)
SELECT p.id, 'affiliate_dash', 10, FALSE
FROM product_catalog p
WHERE p.canonical_sku LIKE 'aff:%' AND p.status = 'active'
ON CONFLICT (product_id, channel) DO NOTHING;
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback)
SELECT p.id, 'manual', 90, TRUE
FROM product_catalog p
WHERE p.canonical_sku LIKE 'aff:%' AND p.status = 'active'
ON CONFLICT (product_id, channel) DO NOTHING;
-- 人工商品:人工优先
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback)
SELECT p.id, 'manual', 10, FALSE
FROM product_catalog p
WHERE p.canonical_sku LIKE 'man:%' AND p.status = 'active'
ON CONFLICT (product_id, channel) DO NOTHING;
-- 卡券:只统计
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback)
SELECT p.id, 'third_party_voucher', 10, FALSE
FROM product_catalog p
WHERE p.canonical_sku LIKE 'tpv:%' AND p.status = 'active'
ON CONFLICT (product_id, channel) DO NOTHING;
-- 接单模板/工单商品
INSERT INTO product_channel_capabilities (product_id, channel, priority, is_fallback)
SELECT p.id, 'work_order', 10, FALSE
FROM product_catalog p
WHERE (p.canonical_sku LIKE 'tpl:%' OR p.canonical_sku LIKE 'wo:%') AND p.status = 'active'
ON CONFLICT (product_id, channel) DO NOTHING;
ALTER TABLE order_items
ADD COLUMN IF NOT EXISTS original_channel TEXT,
ADD COLUMN IF NOT EXISTS channel_switch_reason TEXT NOT NULL DEFAULT '',
ADD COLUMN IF NOT EXISTS channel_switched_at TIMESTAMPTZ;
COMMENT ON COLUMN order_items.original_channel IS
'首次分流渠道;fallback/运营改渠道后仍保留,用于审计';
COMMENT ON COLUMN order_items.channel_switch_reason IS
'渠道切换原因:affiliate_failed / manual_supplement / operator';
ALTER TABLE cost_snapshots
ADD COLUMN IF NOT EXISTS reversed_at TIMESTAMPTZ,
ADD COLUMN IF NOT EXISTS reversal_reason TEXT NOT NULL DEFAULT '';
DROP INDEX IF EXISTS uq_cost_snapshots_active_subject_item;
CREATE UNIQUE INDEX IF NOT EXISTS uq_cost_snapshots_active_subject_item
ON cost_snapshots (subject_type, subject_id, cost_item_id)
WHERE superseded_at IS NULL AND reversed_at IS NULL;
-- 人工成本快照同样排除已冲销行,允许冲销后重新确认。
DROP INDEX IF EXISTS uq_cost_snapshots_manual_order_item;
CREATE UNIQUE INDEX IF NOT EXISTS uq_cost_snapshots_manual_order_item
ON cost_snapshots (order_item_id)
WHERE subject_type = 'manual_case' AND superseded_at IS NULL
AND reversed_at IS NULL AND subject_id > 0;
DO $$
BEGIN
IF EXISTS (
SELECT 1 FROM pg_constraint WHERE conname = 'chk_order_items_fulfillment_source'
) THEN
ALTER TABLE order_items DROP CONSTRAINT chk_order_items_fulfillment_source;
END IF;
ALTER TABLE order_items
ADD CONSTRAINT chk_order_items_fulfillment_source
CHECK (
fulfillment_source IN (
'', 'auto_rule', 'auto_mapping', 'carrier_evidence',
'manual_confirm', 'operator_assigned', 'cutover_archive',
'affiliate_fallback'
)
);
END $$;
@@ -106,11 +106,12 @@ export const WORK_ORDER_TEMPLATE_COST_BASELINE = '2026-07-10T00:00:00+08:00'
*/
export const PRODUCT_SOURCE_PREFIXES: Array<{
prefix: string
source: 'affiliate_dash' | 'work_order' | 'manual'
source: 'affiliate_dash' | 'work_order' | 'third_party_voucher' | 'manual'
}> = [
{ prefix: 'aff:', source: 'affiliate_dash' },
{ prefix: 'tpl:', source: 'work_order' },
{ prefix: 'wo:', source: 'work_order' },
{ prefix: 'tpv:', source: 'third_party_voucher' },
{ prefix: 'man:', source: 'manual' },
]
@@ -298,3 +298,10 @@ test('成本模型与商品来源必须覆盖第三方卡券渠道', () => {
assert.ok(productSourceFilterSql(1).includes("'third_party_voucher'"))
assert.ok(productSourceFilterSql(1).includes("'tpv:%'"))
})
test('tpv: 商品来源归为第三方卡券', async () => {
const { productSourceOf } = await import('./cost-center-mappers.js')
const assert = (await import('node:assert/strict')).default
assert.equal(productSourceOf('tpv:模拟卡券'), 'third_party_voucher')
assert.equal(productSourceOf('man:模拟人工'), 'manual')
})
@@ -295,6 +295,25 @@ export async function resolveExceptionsForItemsByReason(
return result.rowCount || 0
}
/**
* 清理“头还在、明细已空”的孤儿异常。
* 并发收口时可能出现明细已删但头未关闭,按 item 再查也找不到它。
*/
export async function resolveOrphanCostExceptions(): Promise<number> {
const result = await query<{ id: number }>(
`
UPDATE cost_exceptions ce
SET status = 'resolved', resolved_at = NOW(), item_count = 0, last_seen_at = NOW()
WHERE ce.status <> 'resolved'
AND NOT EXISTS (
SELECT 1 FROM cost_exception_items i WHERE i.exception_id = ce.id
)
RETURNING ce.id
`,
)
return result.rowCount || 0
}
export async function resolveExceptionsForItems(orderItemIds: number[]): Promise<number> {
const ids = orderItemIds.filter((id) => Number.isInteger(id) && id > 0)
if (ids.length === 0) return 0
@@ -177,7 +177,7 @@ export async function createAdminCostSnapshot(payload: JsonObject = {}, _session
WHERE subject_type = 'order_item'
AND subject_id = $1
AND cost_item_id = $2
AND superseded_at IS NULL
AND superseded_at IS NULL AND reversed_at IS NULL
FOR UPDATE
`,
[orderItemId, line.costItemId],
@@ -194,7 +194,7 @@ export async function createAdminCostSnapshot(payload: JsonObject = {}, _session
continue
}
await client.query(
`UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL`,
`UPDATE cost_snapshots SET superseded_at = $1 WHERE id = $2 AND superseded_at IS NULL AND reversed_at IS NULL`,
[now, current.id],
)
supersededCount += 1
@@ -212,7 +212,7 @@ export async function createAdminCostSnapshot(payload: JsonObject = {}, _session
$7, $8, $9, $10, 'resolved', $11::jsonb, $12, $13, $13
)
ON CONFLICT (subject_type, subject_id, cost_item_id)
WHERE superseded_at IS NULL
WHERE superseded_at IS NULL AND reversed_at IS NULL
DO NOTHING
RETURNING *
`,
@@ -4,6 +4,7 @@ import { nowIso } from '../../utils/time.js'
import { GENERIC_COST_MODELS } from './cost-center-constants.js'
import {
resolveExceptionsForItems,
resolveExceptionsForItemsByReason,
upsertAggregatedExceptionsBatch,
type AggregatedExceptionEntry,
} from './cost-exception-service.js'
@@ -371,6 +372,7 @@ async function processCostRows(
const inserts: Array<{ item: CostCandidateItem; resolution: NonNullable<CostResolution> }> = []
const supersedeIds: number[] = []
const missingEntries: AggregatedExceptionEntry[] = []
const resolvedMissingIds: number[] = []
for (const item of rows) {
const desired: Array<NonNullable<CostResolution>> = []
@@ -381,15 +383,25 @@ async function processCostRows(
if (channelResolution) {
desired.push(channelResolution)
} else if (code) {
result.missing += 1
// 已在队列里的缺失项不再重复写三条 SQL,等价格补齐后自然收口。
if (!openMissing.has(Number(item.id))) {
missingEntries.push({
reason: 'cost_missing',
scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`,
orderItemId: Number(item.id),
detail: { channel: item.fulfillment_channel },
})
// 历史版本可能已 retired,但订单已有不可变快照:这单不算缺成本,
// 否则“改价后旧订单快照还在、异常却重新冒出来”。
const channelCostItem = ctx.channelCostItems.get(code)
const hasActiveChannelSnapshot = channelCostItem
? existingMap.get(Number(item.id))?.has(Number(channelCostItem.id))
: false
if (hasActiveChannelSnapshot) {
resolvedMissingIds.push(Number(item.id))
} else {
result.missing += 1
// 已在队列里的缺失项不再重复写三条 SQL,等价格补齐后自然收口。
if (!openMissing.has(Number(item.id))) {
missingEntries.push({
reason: 'cost_missing',
scopeKey: `cost_missing|${item.fulfillment_channel}|${item.canonical_product_id}|${item.work_product_rule_id ?? '-'}`,
orderItemId: Number(item.id),
detail: { channel: item.fulfillment_channel },
})
}
}
}
@@ -429,6 +441,11 @@ async function processCostRows(
}
}
if (resolvedMissingIds.length > 0) {
await resolveExceptionsForItemsByReason(resolvedMissingIds, 'cost_missing').catch(
() => undefined,
)
}
if (missingEntries.length > 0) {
await upsertAggregatedExceptionsBatch(missingEntries)
}
@@ -583,7 +600,7 @@ async function loadActiveSnapshots(
`
SELECT id, order_item_id, cost_item_id, cost_version_id, total_amount_fen, snapshot_stage
FROM cost_snapshots
WHERE superseded_at IS NULL
WHERE superseded_at IS NULL AND reversed_at IS NULL
AND order_item_id = ANY($1::bigint[])
`,
[itemIds],
@@ -655,7 +672,7 @@ async function insertSnapshots(
product_id, quantity, unit_amount_fen, total_amount_fen, stage,
channel, basis_at
)
ON CONFLICT (subject_type, subject_id, cost_item_id) WHERE superseded_at IS NULL
ON CONFLICT (subject_type, subject_id, cost_item_id) WHERE superseded_at IS NULL AND reversed_at IS NULL
DO NOTHING
`,
[
@@ -18,6 +18,12 @@ import { resolveSkuNameQuantity } from '../product-match/pricing.js'
import { setAdminAffiliateProductPrice } from './affiliate-product-service.js'
import { resolvePriceEffectiveFrom } from './cost-center-normalizers.js'
import { ensureVoucherRuleForSku } from '../fulfillment/channel-assign-service.js'
import { syncCostSnapshotsForItems } from './cost-snapshot-service.js'
import {
resolveExceptionsForItemsByReason,
resolveOrphanCostExceptions,
} from './cost-exception-service.js'
import { orderItemBasisDates, refreshFactsForDates } from '../finance/refresh-facts.js'
import { syncAdminWorkProductRulePrice } from '../worker-platform/admin-worker-catalog-service.js'
import {
AFFILIATE_TASK_AGG_JOIN,
@@ -1372,7 +1378,7 @@ export async function setAdminProductPrice(
}
const now = nowIso()
const effectiveFrom = resolvePriceEffectiveFrom(payload)
let effectiveFrom = resolvePriceEffectiveFrom(payload)
const changeReason = String(payload.changeReason || '在商品与成本中心统一调整价格').trim()
// 2. 如果是接单模板商品 (tpl:)
@@ -1548,6 +1554,35 @@ export async function setAdminProductPrice(
)
const existing = existingResult.rows[0]
const existingTime = existing ? Date.parse(existing.effective_from) : null
// 首次配置/补价:如果这个商品还有未算成本的订单,默认从最早缺失订单当天生效,
// 否则当天早些时候的订单会因为“版本生效时间晚于下单时间”永远算不到成本。
effectiveFrom = resolvePriceEffectiveFrom(payload)
const requestedEffective = payload.effectiveFrom ?? payload.effective_from
if (!requestedEffective) {
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
JOIN orders o ON o.id = oi.order_id
WHERE oi.canonical_product_id = $1
AND oi.fulfillment_status = 'assigned'
AND NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.order_item_id = oi.id
AND cs.cost_item_id = $2
AND cs.superseded_at IS NULL
AND cs.reversed_at IS NULL
)
`,
[product.id, costItemId],
)
const earliestDate = String(earliest.rows[0]?.basis_date || '').trim()
const earliestTime = earliestDate ? Date.parse(`${earliestDate}T00:00:00+08:00`) : NaN
if (Number.isFinite(earliestTime) && (existingTime === null || earliestTime < existingTime)) {
effectiveFrom = `${earliestDate}T00:00:00+08:00`
}
}
const targetTime = Date.parse(effectiveFrom)
let costVersionId = existing?.id || 0
@@ -1606,6 +1641,59 @@ export async function setAdminProductPrice(
costVersionId = inserted.rows[0]?.id || 0
}
// 配置后立刻补算该商品下缺成本的订单,避免用户看到“已配置但异常还在”。
const missingItems = await query<{ id: number }>(
`
SELECT oi.id
FROM order_items oi
WHERE oi.canonical_product_id = $1
AND oi.fulfillment_status = 'assigned'
AND NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.order_item_id = oi.id
AND cs.cost_item_id = $2
AND cs.superseded_at IS NULL
AND cs.reversed_at IS NULL
)
ORDER BY oi.id
LIMIT 2000
`,
[product.id, costItemId],
)
if (missingItems.rows.length > 0) {
const ids = missingItems.rows.map((row) => Number(row.id))
await syncCostSnapshotsForItems(ids).catch(() => undefined)
const dates = await orderItemBasisDates(ids)
await refreshFactsForDates(dates).catch(() => undefined)
}
// 已有快照但异常还挂着的(并发/历史遗留),保存成本时一并收口。
const coveredItems = await query<{ id: number }>(
`
SELECT oi.id
FROM order_items oi
WHERE oi.canonical_product_id = $1
AND oi.fulfillment_status = 'assigned'
AND EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.order_item_id = oi.id
AND cs.cost_item_id = $2
AND cs.superseded_at IS NULL
AND cs.reversed_at IS NULL
)
ORDER BY oi.id
LIMIT 2000
`,
[product.id, costItemId],
)
if (coveredItems.rows.length > 0) {
await resolveExceptionsForItemsByReason(
coveredItems.rows.map((row) => Number(row.id)),
'cost_missing',
).catch(() => undefined)
}
await resolveOrphanCostExceptions().catch(() => undefined)
return {
success: true,
product,
@@ -1,7 +1,11 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { buildMappingIndex, resolveProductId } from './product-normalize-service.js'
import {
buildMappingIndex,
resolveProductId,
voucherCanonicalSku,
} from './product-normalize-service.js'
const mappings = [
{
@@ -134,3 +138,24 @@ test('商品归一:第三方卡券按 tpv:SKU 直接命中', () => {
41,
)
})
test('三方卡券自动建商品时优先使用内部中文 SKU', () => {
assert.equal(
voucherCanonicalSku({
manual_sku_key: '11月改名卡',
sku_name: '11月改名卡',
sku_code: '192592489694733',
}),
'tpv:11月改名卡',
)
assert.equal(
voucherCanonicalSku({
manual_sku_key: '192592489694733',
sku_name: '11月改名卡',
sku_code: '192592489694733',
}),
'tpv:11月改名卡',
)
assert.equal(voucherCanonicalSku({ manual_sku_key: '', sku_name: '', sku_code: 'x' }), 'tpv:x')
assert.equal(voucherCanonicalSku({}), null)
})
@@ -5,7 +5,8 @@ import { readWatermarkItemId } from '../watermark-store.js'
/**
* 管道 B:商品归一。
*
* 只写 `order_items.canonical_product_id`,不建商品、不改渠道、不碰成本。
* 只写 `order_items.canonical_product_id`,不改渠道、不碰成本;
* 例外:三方卡券缺商品时按内部中文 SKU 自动建 `tpv:` 商品并补渠道规则。
*
* 解析顺序(与旧的四条回填路径统一):
* 1. 自动发货渠道:`aff:<sku_code>` 商品直接命中(该渠道的商品就是 SKU 本身);
@@ -43,6 +44,7 @@ type ItemRow = {
shop_id: string
sku_code: string
sku_name: string
manual_sku_key: string | null
normalized_sku_name: string | null
normalized_sku_series: string | null
fulfillment_channel: string | null
@@ -110,6 +112,94 @@ export function buildMappingIndex(
return index
}
/** 三方卡券的内部 SKU:优先中文,其次名称/平台码。 */
export function pickVoucherInternalSku(item: {
manual_sku_key?: string | null
sku_name?: string | null
sku_code?: string | null
}): string {
const candidates = [item.manual_sku_key, item.sku_name, item.sku_code]
.map((value) => String(value || '').trim())
.filter(Boolean)
return candidates.find((value) => /[\u4e00-\u9fff]/.test(value)) || candidates[0] || ''
}
export function voucherCanonicalSku(item: {
manual_sku_key?: string | null
sku_name?: string | null
sku_code?: string | null
}): string | null {
const internal = pickVoucherInternalSku(item)
return internal ? `tpv:${internal}` : null
}
/**
* 三方卡券没有预先建库时,按内部 SKU 自动建 `tpv:` 商品并补渠道规则。
* 建完后该商品会出现在「商品与成本中心 → 第三方卡券」,运营直接设采购价。
*/
async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null> {
const canonicalSku = voucherCanonicalSku(item)
const internalSku = pickVoucherInternalSku(item)
if (!canonicalSku || !internalSku) return null
const existing = await query<{ id: number; status: string }>(
'SELECT id, status FROM product_catalog WHERE canonical_sku = $1 LIMIT 1',
[canonicalSku],
)
let productId = Number(existing.rows[0]?.id || 0)
if (!productId) {
const created = await query<{ id: number }>(
`
INSERT INTO product_catalog (
canonical_sku, canonical_name, category, unit_name, status,
attributes_json, created_at, updated_at
)
VALUES ($1, $2, '卡券', '张', 'active', '{"source":"third_party_voucher","auto_created":true}'::jsonb, NOW(), NOW())
ON CONFLICT (canonical_sku) DO NOTHING
RETURNING id
`,
[canonicalSku, internalSku],
)
productId = Number(created.rows[0]?.id || 0)
if (!productId) {
const again = await query<{ id: number }>(
'SELECT id FROM product_catalog WHERE canonical_sku = $1 LIMIT 1',
[canonicalSku],
)
productId = Number(again.rows[0]?.id || 0)
}
}
if (!productId) return null
const current = await query<{ status: string }>(
'SELECT status FROM product_catalog WHERE id = $1',
[productId],
)
if (current.rows[0]?.status !== 'active') return null
await query(
'UPDATE order_items SET canonical_product_id = $2, updated_at = NOW() WHERE id = $1 AND canonical_product_id IS NULL',
[item.id, productId],
)
const skuCode = String(item.sku_code || '').trim()
if (skuCode) {
await query(
`
INSERT INTO sku_channel_rules (
provider, shop_id, match_key_type, match_key, channel, priority,
source, enabled, note, created_at, updated_at
)
VALUES ($1, $2, 'sku_code', $3, 'third_party_voucher', 10, 'voucher', TRUE,
'卡券订单自动建商品后生成', NOW(), NOW())
ON CONFLICT (provider, shop_id, match_key_type, match_key, channel) DO NOTHING
`,
[item.provider, item.shop_id, skuCode],
)
}
return productId
}
/** 纯函数:按索引解析商品,便于单测。 */
export function resolveProductId(item: ItemRow, index: MappingIndex): number | null {
const provider = String(item.provider || '').trim()
@@ -204,6 +294,7 @@ export async function normalizeOrderItemProducts(
o.shop_id,
oi.sku_code,
oi.sku_name,
oi.manual_sku_key,
oi.normalized_sku_name,
oi.normalized_sku_series,
oi.fulfillment_channel,
@@ -235,7 +326,10 @@ export async function normalizeOrderItemProducts(
const matched: Array<{ id: number; productId: number }> = []
for (const item of rows.rows) {
const productId = resolveProductId(item, index)
let productId = resolveProductId(item, index)
if (!productId && item.fulfillment_channel === 'third_party_voucher') {
productId = await ensureVoucherProductForItem(item)
}
if (!productId) {
result.unmatched += 1
await upsertAggregatedException({
@@ -338,6 +432,7 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
o.shop_id,
oi.sku_code,
oi.sku_name,
oi.manual_sku_key,
oi.normalized_sku_name,
oi.normalized_sku_series,
oi.fulfillment_channel,
@@ -357,7 +452,10 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
const matched: Array<{ id: number; productId: number }> = []
let unmatched = 0
for (const item of rows.rows) {
const productId = resolveProductId(item, index)
let productId = resolveProductId(item, index)
if (!productId && item.fulfillment_channel === 'third_party_voucher') {
productId = await ensureVoucherProductForItem(item)
}
if (!productId) {
unmatched += 1
await upsertAggregatedException({
@@ -451,7 +451,7 @@ export async function syncAdminWorkOrderTemplateCostVersions(ruleId?: number) {
JOIN rule_version rv ON rv.work_product_rule_id = wo.product_rule_id
WHERE cs.order_item_id = wo.order_item_id
AND cs.cost_item_id = rv.cost_item_id
AND cs.superseded_at IS NULL
AND cs.superseded_at IS NULL AND reversed_at IS NULL
AND cs.cost_version_id <> rv.cost_version_id
AND EXISTS (
SELECT 1 FROM cost_versions gv
@@ -622,7 +622,7 @@ async function assertPeriodCloseable(
SELECT 1 FROM cost_snapshots cs
WHERE cs.subject_type = 'manual_case'
AND cs.order_item_id = oi.id
AND cs.superseded_at IS NULL
AND cs.superseded_at IS NULL AND reversed_at IS NULL
)
`,
[dateFrom, dateTo],
@@ -639,7 +639,7 @@ async function assertPeriodCloseable(
JOIN order_items 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 cs.superseded_at IS NULL AND reversed_at IS NULL
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND cs.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs)
@@ -661,7 +661,7 @@ async function assertPeriodCloseable(
JOIN order_items 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 cs.superseded_at IS NULL AND reversed_at IS NULL
AND (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date
BETWEEN $1::date AND $2::date
AND cs.created_at >= (SELECT MIN(created_at) FROM manual_fulfillment_configs)
@@ -134,7 +134,7 @@ export async function listManualFulfillmentPending(
cs.order_item_id,
MAX(cs.created_at) AS fulfilled_at
FROM cost_snapshots cs
WHERE cs.subject_type = 'manual_case' AND cs.superseded_at IS NULL
WHERE cs.subject_type = 'manual_case' AND cs.superseded_at IS NULL AND reversed_at IS NULL
GROUP BY cs.order_item_id
)
`
@@ -294,7 +294,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
// 锁内复查:并发/双击的第二个请求会等锁,醒来后看到快照直接拒绝,不会重复扣款。
const existing = await client.query<{ id: number }>(
`SELECT id FROM cost_snapshots
WHERE subject_type = 'manual_case' AND order_item_id = $1 AND superseded_at IS NULL
WHERE subject_type = 'manual_case' AND order_item_id = $1 AND superseded_at IS NULL AND reversed_at IS NULL
LIMIT 1`,
[orderItemId],
)
@@ -874,7 +874,7 @@ export async function getManualFulfillmentCountsByOrderIds(
WITH fulfilled AS (
SELECT order_item_id, MAX(created_at) AS fulfilled_at
FROM cost_snapshots
WHERE subject_type = 'manual_case' AND superseded_at IS NULL
WHERE subject_type = 'manual_case' AND superseded_at IS NULL AND reversed_at IS NULL
GROUP BY order_item_id
)
SELECT
@@ -105,18 +105,18 @@ export async function rebuildChannelSalesDailyFacts(input: {
FROM cost_snapshots cs
JOIN cost_items ci ON ci.id = cs.cost_item_id
WHERE cs.order_item_id = s.order_item_id
AND cs.superseded_at IS NULL
AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL
AND ci.code = s.channel_cost_code
)
WHEN s.channel = 'manual' THEN NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.order_item_id = s.order_item_id
AND cs.superseded_at IS NULL
AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL
AND cs.subject_type = 'manual_case'
)
ELSE NOT EXISTS (
SELECT 1 FROM cost_snapshots cs
WHERE cs.order_item_id = s.order_item_id AND cs.superseded_at IS NULL
WHERE cs.order_item_id = s.order_item_id AND cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL
)
END AS cost_missing,
COALESCE(c.has_actual, FALSE) AS has_actual
@@ -128,7 +128,7 @@ export async function rebuildChannelSalesDailyFacts(input: {
SUM(cs.total_amount_fen)::bigint AS cost_fen,
BOOL_AND(cs.snapshot_stage = 'actual') AS has_actual
FROM cost_snapshots cs
WHERE cs.superseded_at IS NULL
WHERE cs.superseded_at IS NULL AND reversed_at IS NULL AND cs.reversed_at IS NULL
GROUP BY cs.order_item_id
) c ON c.order_item_id = s.order_item_id
)
@@ -0,0 +1,12 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { AFFILIATE_FAILURE_STATUSES } from './affiliate-fallback-service.js'
test('自动发货失败状态全部纳入转人工', () => {
for (const status of ['closed', 'expired', 'manual_review', 'retry_pending', 'cancelled']) {
assert.ok(AFFILIATE_FAILURE_STATUSES.includes(status as never))
}
assert.ok(!AFFILIATE_FAILURE_STATUSES.includes('redeemed' as never))
assert.ok(!AFFILIATE_FAILURE_STATUSES.includes('waiting_binding' as never))
})
@@ -0,0 +1,122 @@
import { query, withTransaction } from '../../db/client.js'
import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js'
import { refreshFactsForDates } from '../finance/refresh-facts.js'
/**
* 自动发货失败 → 人工兜底。
*
* 触发:该订单项最新一条 affiliate_dash 任务处于失败状态,且没有任何成功任务。
* 切换后:原渠道成本(sys_affiliate_purchase)冲销,人工成本等确认时再记;
* 自动任务本身不改状态,只记录订单项的切换原因,避免重复触发。
*/
export const AFFILIATE_FAILURE_STATUSES = [
'closed',
'expired',
'manual_review',
'retry_pending',
'cancelled',
] as const
const AFFILIATE_SUCCESS_STATUSES = ['redeemed', 'paid'] as const
export async function runAffiliateFallback(
options: { limit?: number } = {},
): Promise<{ scanned: number; switched: number }> {
const limit = Math.min(1000, Math.max(1, Math.trunc(options.limit || 200)))
const candidates = await query<{
id: number
order_item_id: number
provider: string
shop_id: string
manual_sku_key: string | null
sku_name: string
config_id: number | null
basis_date: string
}>(
`
SELECT
oi.id,
oi.id AS order_item_id,
o.provider,
o.shop_id,
oi.manual_sku_key,
oi.sku_name,
cfg.id AS config_id,
(COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date::text AS basis_date
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
LEFT JOIN manual_fulfillment_configs cfg
ON cfg.sku_key = oi.manual_sku_key AND cfg.enabled
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'affiliate_dash'
AND oi.fulfillment_locked = FALSE
AND EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id
AND ft.executor_key = 'affiliate_dash'
AND ft.task_status = ANY($1::text[])
)
AND NOT EXISTS (
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id
AND ft.executor_key = 'affiliate_dash'
AND ft.task_status = ANY($2::text[])
)
ORDER BY oi.id
LIMIT $3
`,
[AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit],
)
const switchedIds: number[] = []
const affectedDates = new Set<string>()
for (const row of candidates.rows) {
const updated = await withTransaction(async (client) => {
const result = await client.query(
`
UPDATE order_items
SET
original_channel = COALESCE(original_channel, 'affiliate_dash'),
fulfillment_channel = 'manual',
fulfillment_status = 'assigned',
fulfillment_source = 'affiliate_fallback',
channel_switch_reason = 'affiliate_failed',
channel_switched_at = NOW(),
fulfillment_config_id = COALESCE(fulfillment_config_id, $2),
updated_at = NOW()
WHERE id = $1
AND fulfillment_status = 'assigned'
AND fulfillment_channel = 'affiliate_dash'
AND fulfillment_locked = FALSE
RETURNING id
`,
[row.id, row.config_id],
)
if ((result.rowCount || 0) === 0) return false
await client.query(
`
UPDATE cost_snapshots
SET reversed_at = NOW(), reversal_reason = 'affiliate_failed'
WHERE order_item_id = $1
AND cost_item_id = (SELECT id FROM cost_items WHERE code = 'sys_affiliate_purchase')
AND superseded_at IS NULL AND reversed_at IS NULL
AND reversed_at IS NULL
`,
[row.id],
)
return true
})
if (!updated) continue
switchedIds.push(Number(row.id))
affectedDates.add(row.basis_date)
}
for (const id of switchedIds) {
await syncCostSnapshotsForItems([id]).catch(() => undefined)
}
await refreshFactsForDates([...affectedDates]).catch(() => undefined)
return { scanned: candidates.rows.length, switched: switchedIds.length }
}
@@ -1,8 +1,10 @@
import type { JsonObject } from '../../types/json.js'
import { query } from '../../db/client.js'
import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
import { runAffiliateFallback } from '../fulfillment/affiliate-fallback-service.js'
import {
resolveExceptionsForItemsByReason,
resolveOrphanCostExceptions,
upsertAggregatedExceptionsBatch,
type AggregatedExceptionEntry,
} from '../cost-center/cost-exception-service.js'
@@ -167,7 +169,7 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
SELECT 1 FROM cost_snapshots cs
WHERE cs.subject_type = 'manual_case'
AND cs.order_item_id = oi.id
AND cs.superseded_at IS NULL
AND cs.superseded_at IS NULL AND reversed_at IS NULL
)
AND NOT EXISTS (
SELECT 1
@@ -252,17 +254,23 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
const resolved = await resolveCarrierDispatchExceptions()
const fallback = await runAffiliateFallback({ limit: 200 }).catch(() => ({
scanned: 0,
switched: 0,
}))
const nextCursor = maxId === 0 ? 0 : maxId
await touchWatermark('carrier_dispatch', { lastItemId: nextCursor })
return {
status: 'ok',
message:
`人工载体挂载 ${attached},缺配置 ${missingManual.rows.length},` +
`缺履约载体 ${missingCarrier.rows.length},异常收口 ${resolved}`,
`缺履约载体 ${missingCarrier.rows.length},异常收口 ${resolved},` +
`自动转人工 ${fallback.switched}`,
attached,
missingManual: missingManual.rows.length,
missingCarrier: missingCarrier.rows.length,
resolved,
affiliateFallback: fallback.switched,
}
}
@@ -335,7 +343,7 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
SELECT 1 FROM cost_snapshots cs
WHERE cs.subject_type = 'manual_case'
AND cs.order_item_id = i.order_item_id
AND cs.superseded_at IS NULL
AND cs.superseded_at IS NULL AND reversed_at IS NULL
)
LIMIT 500
`,
@@ -383,6 +391,7 @@ export async function runCostSnapshotJob(job: JsonObject = {}) {
await touchWatermark('cost_snapshot', { lastItemId: result.nextCursor })
await touchWatermark('cost_snapshot_hot', { lastItemId: result.hotCursor })
const refreshed = await refreshFactsForDates(result.affectedDates)
await resolveOrphanCostExceptions().catch(() => undefined)
return {
status: 'ok',
message: