Merge branch 'main' into dev_match

This commit is contained in:
2026-11-06 10:14:43 +08:00
63 changed files with 7027 additions and 954 deletions
@@ -348,6 +348,7 @@ export async function getAdminTasks(
roleId: String(query.roleId || '').trim(),
dateFrom: normalizeDateQuery(query.dateFrom),
dateTo: normalizeDateQuery(query.dateTo, true),
currentOnly: String(query.currentOnly ?? 'true') !== 'false',
})
const viewerContext = createAdminViewerContext(session)
@@ -7,16 +7,22 @@ import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js
import { nowIso } from '../../utils/time.js'
import { rebuildChannelSalesDailyFacts } from '../finance/sales-facts-channel-service.js'
import { AFFILIATE_DASH_PRICES_BASELINE } from './affiliate-dash-prices-baseline.js'
import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
/**
* 清理历史上因为把商品中文名误当成 SKU 插入的非标脏商品,并将误关联的订单项重新归一到标准干净商品。
*/
export async function cleanupDirtyAffiliateDashProducts() {
await query(`
const dirtyMappings = await query<{
order_item_id: number
clean_product_id: number
expected_product_id: number
}>(`
WITH dirty_map AS (
SELECT DISTINCT ON (oi.id)
oi.id AS order_item_id,
clean.id AS clean_product_id
clean.id AS clean_product_id,
dirty.id AS expected_product_id
FROM order_items 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'
@@ -27,11 +33,20 @@ export async function cleanupDirtyAffiliateDashProducts() {
WHERE dirty.canonical_sku ~ '^aff:[^a-z0-9_]'
ORDER BY oi.id, ft.id DESC
)
UPDATE order_items oi
SET canonical_product_id = dm.clean_product_id
FROM dirty_map dm
WHERE oi.id = dm.order_item_id
SELECT order_item_id, clean_product_id, expected_product_id
FROM dirty_map
ORDER BY order_item_id
`)
if (dirtyMappings.rows.length > 0) {
await attachFulfillmentConfigurations({
mappings: dirtyMappings.rows.map((row) => ({
orderItemId: Number(row.order_item_id),
canonicalProductId: Number(row.clean_product_id),
expectedCanonicalProductId: Number(row.expected_product_id),
})),
onlyWhenCanonicalProductDifferent: true,
})
}
await query(`
DELETE FROM sales_daily_facts
@@ -134,8 +149,6 @@ export async function syncAdminAffiliateDashProducts() {
type AffiliateSkuBackfillRow = {
affiliate_item_count: number
matched_count: number
updated_count: number
from_date: string | null
to_date: string | null
}
@@ -165,8 +178,11 @@ export async function rebuildFactsForRange(fromDate: string, toDate: string) {
* 不写 channel_product_mappings。多个任务共用同一订单项时取最新 SKU。
*/
export async function backfillAdminAffiliateDashSkus() {
const now = nowIso()
const result = await query<AffiliateSkuBackfillRow>(
const targets = await query<{
id: number
canonical_product_id: number
expected_product_id: number | null
}>(
`
WITH affiliate_item AS (
SELECT DISTINCT ON (ft.order_item_id)
@@ -184,30 +200,37 @@ export async function backfillAdminAffiliateDashSkus() {
ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
),
target AS (
SELECT ai.id, p.id AS canonical_product_id
SELECT
ai.id,
p.id AS canonical_product_id,
oi.canonical_product_id AS expected_product_id
FROM affiliate_item ai
JOIN product_catalog p
ON p.canonical_sku = 'aff:' || ai.sku_code
),
updated AS (
UPDATE order_items oi
SET canonical_product_id = target.canonical_product_id,
updated_at = $1
FROM target
WHERE oi.id = target.id
AND oi.canonical_product_id IS DISTINCT FROM target.canonical_product_id
RETURNING oi.id
JOIN order_items oi ON oi.id = ai.id
)
SELECT id, canonical_product_id, expected_product_id
FROM target
ORDER BY id
`,
)
const result = await query<AffiliateSkuBackfillRow>(
`
WITH affiliate_item AS (
SELECT DISTINCT ON (ft.order_item_id)
ft.order_item_id AS id
FROM fulfillment_tasks ft
WHERE ft.executor_key = 'affiliate_dash'
AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
AND ft.created_at >= NOW() - INTERVAL '30 days'
AND btrim(COALESCE(ft.context_json->>'skuCode', '')) <> ''
ORDER BY ft.order_item_id, ft.created_at DESC, ft.id DESC
)
SELECT
(
SELECT COUNT(DISTINCT ft.order_item_id)::int
FROM fulfillment_tasks ft
WHERE ft.executor_key = 'affiliate_dash'
AND ft.task_status NOT IN ('closed','expired','failed','manual_review','retry_pending','cancelled')
AND ft.created_at >= NOW() - INTERVAL '30 days'
SELECT COUNT(*)::int
FROM affiliate_item
) AS affiliate_item_count,
(SELECT COUNT(*)::int FROM target) AS matched_count,
(SELECT COUNT(*)::int FROM updated) AS updated_count,
(
SELECT to_char(
GREATEST(
@@ -230,13 +253,24 @@ export async function backfillAdminAffiliateDashSkus() {
JOIN orders o ON o.id = oi.order_id
) AS to_date
`,
[now],
)
const attached =
targets.rows.length > 0
? await attachFulfillmentConfigurations({
mappings: targets.rows.map((row) => ({
orderItemId: Number(row.id),
canonicalProductId: Number(row.canonical_product_id),
expectedCanonicalProductId:
row.expected_product_id === null ? null : Number(row.expected_product_id),
})),
onlyWhenCanonicalProductDifferent: true,
})
: []
const row = result.rows[0]
const affiliateItemCount = Number(row?.affiliate_item_count || 0)
const matchedCount = Number(row?.matched_count || 0)
const updatedCount = Number(row?.updated_count || 0)
const matchedCount = targets.rows.length
const updatedCount = attached.length
const skippedCount = Math.max(0, matchedCount - updatedCount)
const remainingResult = await query<{ total: number }>(
`
@@ -291,6 +291,25 @@ test('成本模型与商品来源必须覆盖第三方卡券渠道', () => {
assert.ok(productSourceFilterSql(1).includes("'tpv:%'"))
})
test('第三方卡券删除:无历史引用硬删除,有历史引用自动归档并停用规则', async () => {
const { readFile } = await import('node:fs/promises')
const [serviceSource, routeSource] = await Promise.all([
readFile(new URL('./product-catalog-service.ts', import.meta.url), 'utf8'),
readFile(new URL('../../routes/admin/cost-center-catalog.ts', import.meta.url), 'utf8'),
])
const start = serviceSource.indexOf('export async function deleteAdminVoucherProduct')
const end = serviceSource.indexOf('/**\n * 批量归档旧人工归一商品', start)
const body = serviceSource.slice(start, end)
assert.ok(body.includes('withTransaction'), '卡券删除必须在事务内完成')
assert.ok(body.includes("action: 'archived'"), '有历史引用时必须自动归档')
assert.ok(body.includes("status = 'archived'"), '归档必须保留商品和历史成本')
assert.ok(body.includes('UPDATE channel_product_mappings'), '归档后必须停用旧映射,避免继续命中')
assert.ok(body.includes('DELETE FROM cost_versions'), '无历史引用时应清理采购价版本')
assert.ok(body.includes("channel = 'third_party_voucher'"), '必须停用对应卡券渠道规则')
assert.ok(routeSource.includes("'/cost-center/voucher-products/:productId'"))
})
test('人工客服筛选排除已有其他主渠道的旧 man 商品', () => {
const sql = productSourceFilterSql(1)
assert.ok(sql.includes(`$1 = 'manual'`))
@@ -6,6 +6,11 @@ import { formatShanghaiDate, shiftDate } from '../../utils/finance-date-range.js
import { nowIso } from '../../utils/time.js'
import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js'
import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
import {
attachFulfillmentConfigurations,
transitionFulfillmentStateBatch,
transitionFulfillmentStateBatchWithExecutor,
} from '../fulfillment/fulfillment-state-service.js'
import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js'
import { refreshFactsForDates } from '../finance/refresh-facts.js'
import {
@@ -13,6 +18,7 @@ import {
normalizeManualCreditQuantity,
parseManualChannelQuantity,
} from './manual-credit-quantity.js'
import { productSourceFilterSql } from './cost-center-mappers.js'
/**
* 人工履约 SKU 配置服务。
@@ -223,9 +229,7 @@ export async function scanManualSkus(options: ScanOptions) {
return { scanned: rows.length, registered: rows.length, fromDate, toDate }
}
/**
* 列出人工 SKU 清单(登记表 + 配置状态)。
*/
/** 列出人工标准商品清单(标准商品为唯一主键,配置与订单统计只做关联)。 */
export async function listManualSkus(input: Record<string, unknown> = {}) {
const unitPrices = await loadPlatformUnitPriceFen()
const page = normalizePositiveInteger(input.page, 1, '页码')
@@ -238,13 +242,16 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
const platformKey = String(input.platformKey || input.platform_key || '').trim()
const provider = String(input.provider || '').trim()
const conditions: string[] = ['1 = 1']
const params: unknown[] = []
const conditions: string[] = [`p.status = 'active'`, `p.canonical_sku LIKE 'man:%'`]
const params: unknown[] = ['manual']
conditions.push(productSourceFilterSql(params.length))
if (keyword) {
params.push(`%${escapeLike(keyword)}%`)
const placeholder = `$${params.length}`
conditions.push(`(r.sku_key ILIKE ${placeholder} OR r.display_name ILIKE ${placeholder})`)
conditions.push(
`(p.canonical_sku ILIKE ${placeholder} OR p.canonical_name ILIKE ${placeholder})`,
)
}
if (status === 'configured') {
conditions.push('cfg.id IS NOT NULL')
@@ -266,7 +273,7 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
WHERE oi.fulfillment_status = 'assigned' AND oi.fulfillment_channel = 'manual'
AND o.provider = $${params.length}
AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code)
= canonicalize_internal_sku(r.sku_key)
= canonicalize_internal_sku(substring(p.canonical_sku from 5))
)`)
}
@@ -274,9 +281,15 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
const countResult = await query<{ total: number }>(
`
SELECT COUNT(*)::int AS total
FROM manual_sku_registry r
LEFT JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = canonicalize_internal_sku(r.sku_key)
FROM product_catalog p
LEFT JOIN LATERAL (
SELECT *
FROM manual_fulfillment_configs cfg
WHERE canonicalize_internal_sku(cfg.sku_key)
= canonicalize_internal_sku(substring(p.canonical_sku from 5))
ORDER BY cfg.id
LIMIT 1
) cfg ON TRUE
WHERE ${where}
`,
params,
@@ -284,22 +297,56 @@ export async function listManualSkus(input: Record<string, unknown> = {}) {
const listParams = [...params, pageSize, (page - 1) * pageSize]
const result = await query<ManualSkuRow>(
`
WITH manual_stats AS (
SELECT
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key,
COUNT(DISTINCT o.shop_id)::int AS shop_count,
COUNT(DISTINCT o.id)::int AS order_count,
COUNT(*)::int AS item_count,
COALESCE(SUM(GREATEST(COALESCE(oi.revenue_amount_fen, 0), 0)), 0)::bigint
AS revenue_fen,
MIN((COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date)
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
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
GROUP BY 1
)
SELECT
r.id, r.sku_key, r.display_name, r.shop_count, r.order_count, r.item_count,
r.revenue_fen, r.first_seen_date, r.last_seen_date,
r.scan_from_date, r.scan_to_date,
p.id,
substring(p.canonical_sku from 5) AS sku_key,
p.canonical_name AS display_name,
COALESCE(manual_stats.shop_count, 0)::int AS shop_count,
COALESCE(manual_stats.order_count, 0)::int AS order_count,
COALESCE(manual_stats.item_count, 0)::int AS item_count,
COALESCE(manual_stats.revenue_fen, 0)::bigint AS revenue_fen,
manual_stats.first_seen_date,
manual_stats.last_seen_date,
manual_stats.first_seen_date AS scan_from_date,
manual_stats.last_seen_date AS scan_to_date,
cfg.id AS config_id,
cfg.platform_key AS config_platform_key,
cfg.credit_quantity AS config_credit_quantity,
cfg.cost_mode AS config_cost_mode,
cfg.canonical_product_id AS config_canonical_product_id,
COALESCE(cfg.canonical_product_id, p.id) AS config_canonical_product_id,
cfg.enabled AS config_enabled,
cfg.channels_json AS config_channels_json
FROM manual_sku_registry r
LEFT JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = canonicalize_internal_sku(r.sku_key)
FROM product_catalog p
LEFT JOIN LATERAL (
SELECT *
FROM manual_fulfillment_configs cfg
WHERE canonicalize_internal_sku(cfg.sku_key)
= canonicalize_internal_sku(substring(p.canonical_sku from 5))
ORDER BY cfg.id
LIMIT 1
) cfg ON TRUE
LEFT JOIN manual_stats
ON manual_stats.sku_key = canonicalize_internal_sku(substring(p.canonical_sku from 5))
WHERE ${where}
ORDER BY r.item_count DESC, r.sku_key ASC
ORDER BY COALESCE(manual_stats.item_count, 0) DESC, p.canonical_name ASC, p.id ASC
LIMIT $${listParams.length - 1} OFFSET $${listParams.length}
`,
listParams,
@@ -323,20 +370,50 @@ export async function getManualSkuOverview() {
last_scan_at: string | null
}>(
`
WITH manual_stats AS (
SELECT
resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) AS sku_key,
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
JOIN orders o ON o.id = oi.order_id
WHERE oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
GROUP BY 1
),
manual_scope AS (
SELECT p.id, p.canonical_sku, p.canonical_name, cfg.id AS config_id
FROM product_catalog p
LEFT JOIN LATERAL (
SELECT cfg.id
FROM manual_fulfillment_configs cfg
WHERE canonicalize_internal_sku(cfg.sku_key)
= canonicalize_internal_sku(substring(p.canonical_sku from 5))
ORDER BY cfg.id
LIMIT 1
) cfg ON TRUE
WHERE p.status = 'active'
AND p.canonical_sku LIKE 'man:%'
AND ${productSourceFilterSql(1)}
)
SELECT
COUNT(*)::int AS total_sku,
COUNT(*) FILTER (WHERE cfg.id IS NOT NULL)::int AS configured_sku,
COUNT(*) FILTER (WHERE cfg.id IS NULL)::int AS unconfigured_sku,
COALESCE(SUM(r.item_count), 0)::bigint AS total_items,
COALESCE(SUM(r.item_count) FILTER (WHERE cfg.id IS NOT NULL), 0)::bigint
COUNT(*) FILTER (WHERE config_id IS NOT NULL)::int AS configured_sku,
COUNT(*) FILTER (WHERE config_id IS NULL)::int AS unconfigured_sku,
COALESCE(SUM(manual_stats.item_count), 0)::bigint AS total_items,
COALESCE(SUM(manual_stats.item_count) FILTER (WHERE config_id IS NOT NULL), 0)::bigint
AS configured_items,
COALESCE(SUM(r.item_count) FILTER (WHERE cfg.id IS NULL), 0)::bigint
COALESCE(SUM(manual_stats.item_count) FILTER (WHERE config_id IS NULL), 0)::bigint
AS unconfigured_items,
MAX(r.updated_at) AS last_scan_at
FROM manual_sku_registry r
LEFT JOIN manual_fulfillment_configs cfg
ON canonicalize_internal_sku(cfg.sku_key) = canonicalize_internal_sku(r.sku_key)
MAX(manual_stats.last_seen_date)::text AS last_scan_at
FROM manual_scope
LEFT JOIN manual_stats
ON manual_stats.sku_key = canonicalize_internal_sku(
substring(manual_scope.canonical_sku from 5)
)
`,
['manual'],
)
const row = result.rows[0]
return {
@@ -470,6 +547,7 @@ export async function saveManualFulfillmentConfig(payload: JsonObject) {
})
await syncRegistryConfigId(config.sku_key, config.id)
await upsertManualRegistryConfig(config)
// 同步渠道规则:人工配置就是「这个内部 SKU 走人工渠道」的唯一声明,
// 少了这一步,新配的 SKU 会一直停在待分类。
await syncManualChannelRule(config)
@@ -508,12 +586,13 @@ async function retryPendingManualItems(skuKey: string): Promise<number> {
)
const ids = candidates.rows.map((row) => Number(row.id))
if (ids.length === 0) return 0
await query(
`UPDATE order_items
SET fulfillment_locked = FALSE, updated_at = NOW()
WHERE id = ANY($1::bigint[]) AND fulfillment_status = 'pending'`,
[ids],
)
await transitionFulfillmentStateBatch({
orderItemIds: ids,
expected: { status: 'pending', locked: true },
patch: { locked: false },
eventType: 'manual_config_retry_unlocked',
actor: { type: 'system' },
})
const result = await reconcilePendingCarriers({
orderItemIds: ids,
limit: ids.length,
@@ -556,6 +635,7 @@ export type DeleteManualSkuResult = {
deletedConfigs: number
deletedRegistry: number
deletedRules: number
archivedProducts: number
refreshedDays: number
skippedClosedDays: number
}
@@ -671,31 +751,46 @@ export async function deleteManualSkuClassification(
RETURNING id`,
[key],
)
const products = await client.query(
`
UPDATE product_catalog
SET status = 'archived',
attributes_json = attributes_json || jsonb_build_object(
'deletedAt', $2::text,
'deletedBy', 'manual_sku_config'
),
updated_at = $2::timestamptz
WHERE status = 'active'
AND canonical_sku LIKE 'man:%'
AND canonicalize_internal_sku(substring(canonical_sku from 5)) = $1
RETURNING id
`,
[key, nowIso()],
)
if (itemIds.length > 0) {
await client.query(
`
UPDATE order_items
SET fulfillment_channel = NULL,
fulfillment_status = 'pending',
fulfillment_source = '',
fulfillment_locked = FALSE,
fulfillment_config_id = NULL,
fulfillment_pending_reason = 'no_channel_capability',
fulfillment_pending_detail = jsonb_build_object(
'reason', 'manual_classification_removed',
'removedAt', NOW()
),
fulfillment_reason = '运营移除错误的人工分类',
fulfillment_decided_by = NULL,
fulfillment_decided_at = NULL,
updated_at = NOW()
WHERE id = ANY($1::bigint[])
AND fulfillment_status = 'assigned'
AND fulfillment_channel = 'manual'
`,
[itemIds],
)
await transitionFulfillmentStateBatchWithExecutor(client, {
orderItemIds: itemIds,
expected: { status: 'assigned', channel: 'manual' },
patch: {
status: 'pending',
channel: null,
source: '',
locked: false,
configId: null,
pendingReason: 'no_channel_capability',
pendingDetail: {
reason: 'manual_classification_removed',
removedAt: nowIso(),
},
reason: '运营移除错误的人工分类',
decidedBy: null,
decidedAt: null,
},
eventType: 'manual_classification_removed',
actor: { type: 'operator' },
reason: key,
})
}
return {
@@ -703,6 +798,7 @@ export async function deleteManualSkuClassification(
deletedConfigs: configs.rowCount || 0,
deletedRegistry: registry.rowCount || 0,
deletedRules: rules.rowCount || 0,
archivedProducts: products.rowCount || 0,
dates: [...new Set(items.rows.map((row) => row.basis_date).filter(Boolean))],
}
})
@@ -714,6 +810,7 @@ export async function deleteManualSkuClassification(
deletedConfigs: result.deletedConfigs,
deletedRegistry: result.deletedRegistry,
deletedRules: result.deletedRules,
archivedProducts: result.archivedProducts,
refreshedDays: refreshed.days,
skippedClosedDays: refreshed.skippedClosed,
}
@@ -731,11 +828,14 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
const dateFrom = normalizeOptionalDate(input.dateFrom ?? input.date_from, '开始日期')
const dateTo = normalizeOptionalDate(input.dateTo ?? input.date_to, '结束日期')
const now = nowIso()
const result = await query<{ id: number }>(
const candidates = await query<{
order_item_id: number
config_id: number
canonical_product_id: number | null
}>(
`
WITH target AS (
SELECT
SELECT DISTINCT ON (oi.id)
oi.id AS order_item_id,
cfg.id AS config_id,
cfg.canonical_product_id
@@ -757,20 +857,27 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
$4::date IS NULL
OR (COALESCE(o.paid_at, o.created_at) AT TIME ZONE 'Asia/Shanghai')::date <= $4::date
)
ORDER BY oi.id, cfg.id
)
UPDATE order_items oi
SET fulfillment_config_id = target.config_id,
canonical_product_id = COALESCE(target.canonical_product_id, oi.canonical_product_id),
updated_at = $5
SELECT order_item_id, config_id, canonical_product_id
FROM target
WHERE oi.id = target.order_item_id
RETURNING oi.id
ORDER BY order_item_id
`,
[skuKey, onlyUnmatched, dateFrom, dateTo, now],
[skuKey, onlyUnmatched, dateFrom, dateTo],
)
const attached = await attachFulfillmentConfigurations({
mappings: candidates.rows.map((row) => ({
orderItemId: Number(row.order_item_id),
configId: Number(row.config_id),
...(row.canonical_product_id ? { canonicalProductId: Number(row.canonical_product_id) } : {}),
})),
expected: { status: 'assigned', channel: 'manual' },
onlyWhenConfigNull: onlyUnmatched,
onlyWhenCanonicalProductNull: true,
})
// 配置补上以后,「人工 SKU 未配置」异常要立即收口,不能等人工去点。
const updatedIds = result.rows.map((row) => Number(row.id))
const updatedIds = attached.map((row) => Number(row.id))
if (updatedIds.length > 0) {
await resolveExceptionsForItemsByReason(updatedIds, 'manual_config_missing')
}
@@ -779,7 +886,7 @@ export async function rematchManualFulfillments(input: Record<string, unknown> =
await attachCostVersionsForManualConfigs()
return {
updatedCount: result.rowCount || 0,
updatedCount: attached.length,
skuKey,
dateFrom,
dateTo,
@@ -840,6 +947,40 @@ async function syncRegistryConfigId(skuKey: string, configId: number | null) {
)
}
/** 配置保存后确保登记表仍有同一 SKU 的记录,兼容旧扫描清单和运维查询。 */
async function upsertManualRegistryConfig(config: {
id: number
sku_key: string
canonical_product_id: number | null
}) {
const key = canonicalizeInternalSku(config.sku_key)
if (!key) return
const now = nowIso()
await query(
`
INSERT INTO manual_sku_registry (
sku_key, display_name, config_id, created_at, updated_at
)
SELECT
$1,
COALESCE(NULLIF(product.canonical_name, ''), $1),
$2,
$3,
$3
FROM (SELECT 1) seed
LEFT JOIN product_catalog product ON product.id = $4
ON CONFLICT (sku_key) DO UPDATE SET
display_name = CASE
WHEN BTRIM(manual_sku_registry.display_name) = '' THEN EXCLUDED.display_name
ELSE manual_sku_registry.display_name
END,
config_id = EXCLUDED.config_id,
updated_at = EXCLUDED.updated_at
`,
[key, config.id, now, config.canonical_product_id],
)
}
/** 读取单个 SKU 的配置详情(含平台标签与成本折算)。 */
export async function getManualFulfillmentConfig(skuKey: string) {
const unitPrices = await loadPlatformUnitPriceFen()
@@ -19,6 +19,7 @@ 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 { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
import {
resolveExceptionsForItemsByReason,
resolveOrphanCostExceptions,
@@ -439,6 +440,152 @@ export async function deleteAdminProduct(productId: number | string) {
return { deleted: true, product: mapProduct(current) }
}
/**
* 删除第三方卡券商品。
*
* 卡券商品通常带采购价版本,不能套用普通商品“有任何成本版本就拒绝”的规则:
* - 没有任何订单项/映射/成本快照:删除未使用成本版本、渠道能力与卡券规则后硬删除;
* - 已有历史引用:归档商品并停用后续卡券规则,成本版本与快照全部保留。
*/
export async function deleteAdminVoucherProduct(productId: number | string) {
const id = normalizeId(productId, '商品 ID')
return withTransaction(async (client) => {
const currentResult = await client.query<ProductRow>(
`SELECT * FROM product_catalog WHERE id = $1 FOR UPDATE`,
[id],
)
const current = requiredRow(currentResult.rows[0], '商品不存在')
if (!current.canonical_sku.startsWith('tpv:')) {
throw createHttpError('该商品不是第三方卡券商品', {
statusCode: 400,
errorCode: 'voucher_product_type_invalid',
})
}
const usageResult = await client.query<{
mapping_count: number
snapshot_count: number
order_item_count: number
}>(
`
SELECT
(SELECT COUNT(*)::int FROM channel_product_mappings WHERE canonical_product_id = $1)
AS mapping_count,
(
SELECT COUNT(*)::int
FROM cost_snapshots
WHERE canonical_product_id = $1
OR cost_version_id IN (
SELECT id FROM cost_versions WHERE canonical_product_id = $1
)
)
AS snapshot_count,
(SELECT COUNT(*)::int FROM order_items WHERE canonical_product_id = $1)
AS order_item_count
`,
[id],
)
const usage = usageResult.rows[0]
const mappingCount = toNumber(usage?.mapping_count)
const snapshotCount = toNumber(usage?.snapshot_count)
const orderItemCount = toNumber(usage?.order_item_count)
const internalSku = current.canonical_sku.slice(4)
await client.query(
`
UPDATE internal_sku_channels
SET enabled = FALSE,
is_fallback = FALSE,
updated_at = $3
WHERE channel = 'third_party_voucher'
AND (
canonical_product_id = $1
OR canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($2)
)
`,
[id, internalSku, nowIso()],
)
await client.query(
`
UPDATE product_channel_capabilities
SET enabled = FALSE,
updated_at = $2
WHERE product_id = $1 AND channel = 'third_party_voucher'
`,
[id, nowIso()],
)
if (mappingCount > 0 || snapshotCount > 0 || orderItemCount > 0) {
await client.query(
`
UPDATE channel_product_mappings
SET enabled = FALSE,
note = CASE
WHEN note = '' THEN '第三方卡券商品删除后自动停用'
ELSE note
END,
updated_at = $2
WHERE canonical_product_id = $1
`,
[id, nowIso()],
)
const archived = await client.query<ProductRow>(
`
UPDATE product_catalog
SET status = 'archived',
attributes_json = attributes_json || jsonb_build_object(
'deletedAt', $2::text,
'deletedBy', 'voucher_product_panel'
),
updated_at = $2::timestamptz
WHERE id = $1
RETURNING *
`,
[id, nowIso()],
)
return {
action: 'archived' as const,
deleted: false,
product: mapProduct(requiredRow(archived.rows[0], '归档卡券商品失败')),
preserved: {
mappingCount,
snapshotCount,
orderItemCount,
},
}
}
const deletedVersions = await client.query(
`DELETE FROM cost_versions WHERE canonical_product_id = $1`,
[id],
)
await client.query(`DELETE FROM product_channel_capabilities WHERE product_id = $1`, [id])
await client.query(
`
DELETE FROM internal_sku_channels
WHERE channel = 'third_party_voucher'
AND (
canonical_product_id = $1
OR canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($2)
)
`,
[id, internalSku],
)
await client.query('DELETE FROM product_catalog WHERE id = $1', [id])
return {
action: 'deleted' as const,
deleted: true,
product: mapProduct(current),
deletedCostVersionCount: deletedVersions.rowCount || 0,
preserved: {
mappingCount: 0,
snapshotCount: 0,
orderItemCount: 0,
},
}
})
}
/**
* 批量归档旧人工归一商品。
*
@@ -1229,19 +1376,14 @@ export async function backfillAdminProductMappings(payload: JsonObject = {}) {
}
if (ids.length > 0) {
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = NOW()
FROM unnest($1::bigint[], $2::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS NULL
`,
[ids, productIds],
)
updatedCount += updated.rowCount || 0
const updated = await attachFulfillmentConfigurations({
mappings: ids.map((orderItemId, index) => ({
orderItemId,
canonicalProductId: productIds[index]!,
})),
onlyWhenCanonicalProductNull: true,
})
updatedCount += updated.length
}
batches += 1
if (!all || itemsResult.rows.length < limit) break
@@ -3,6 +3,7 @@ import { resolveExceptionsForItems, upsertAggregatedException } from './cost-exc
import { readWatermarkItemId } from '../watermark-store.js'
import { upsertInternalSkuChannel } from '../fulfillment/internal-sku-channel-service.js'
import { canonicalizeInternalSku, containsCjk } from '../fulfillment/internal-sku-resolver.js'
import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
/**
* 管道 B:商品归一。
@@ -188,10 +189,10 @@ async function ensureVoucherProductForItem(item: ItemRow): Promise<number | null
)
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],
)
await attachFulfillmentConfigurations({
mappings: [{ orderItemId: Number(item.id), canonicalProductId: productId }],
onlyWhenCanonicalProductNull: true,
})
await query(
`
@@ -368,17 +369,15 @@ export async function normalizeOrderItemProducts(
}
if (matched.length > 0) {
await query(
`
UPDATE order_items oi
SET canonical_product_id = input.product_id, updated_at = NOW()
FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS NULL
`,
[matched.map((row) => row.id), matched.map((row) => row.productId)],
)
await resolveExceptionsForItems(matched.map((row) => row.id))
const attached = await attachFulfillmentConfigurations({
mappings: matched.map((row) => ({
orderItemId: row.id,
canonicalProductId: row.productId,
})),
expected: { status: 'assigned' },
onlyWhenCanonicalProductNull: true,
})
await resolveExceptionsForItems(attached.map((row) => Number(row.id)))
}
}
@@ -508,17 +507,16 @@ export async function normalizeProductsForItems(itemIds: number[]): Promise<{
matched.push({ id: item.id, productId })
}
if (matched.length > 0) {
await query(
`
UPDATE order_items oi
SET canonical_product_id = input.product_id, updated_at = NOW()
FROM unnest($1::bigint[], $2::bigint[]) AS input(id, product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS NULL
`,
[matched.map((row) => row.id), matched.map((row) => row.productId)],
)
await resolveExceptionsForItems(matched.map((row) => row.id))
const attached = await attachFulfillmentConfigurations({
mappings: matched.map((row) => ({
orderItemId: row.id,
canonicalProductId: row.productId,
})),
expected: { status: 'assigned' },
onlyWhenCanonicalProductNull: true,
})
await resolveExceptionsForItems(attached.map((row) => Number(row.id)))
return { matched: attached.length, unmatched }
}
return { matched: matched.length, unmatched }
return { matched: 0, unmatched }
}
@@ -22,6 +22,7 @@ import {
buildTemplateProductSku,
inferTemplateUnitName,
} from './work-order-template-product-service.js'
import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
export { buildTemplateProductSku, inferTemplateUnitName }
@@ -621,10 +622,10 @@ function compareMappingCandidates(left: TemplateMappingCandidate, right: Templat
async function resetImportedTemplateProductBootstrap(range: { fromDate: string; toDate: string }) {
let resetOrderItemCount = 0
for (let batch = 0; batch < 20; batch += 1) {
const result = await query<{ id: number }>(
const candidates = await query<{ id: number; expected_product_id: number }>(
`
WITH candidates AS (
SELECT oi.id
SELECT oi.id, oi.canonical_product_id AS expected_product_id
FROM order_items oi
JOIN orders o ON o.id = oi.order_id
JOIN channel_product_mappings m
@@ -655,17 +656,22 @@ async function resetImportedTemplateProductBootstrap(range: { fromDate: string;
BETWEEN $1::date AND $2::date
LIMIT 20000
)
UPDATE order_items oi
SET canonical_product_id = NULL,
updated_at = NOW()
FROM candidates c
WHERE oi.id = c.id
RETURNING oi.id
SELECT DISTINCT id, expected_product_id
FROM candidates
ORDER BY id
`,
[range.fromDate, range.toDate],
)
resetOrderItemCount += result.rowCount || 0
if ((result.rowCount || 0) < 20000) break
if (candidates.rows.length === 0) break
const reset = await attachFulfillmentConfigurations({
mappings: candidates.rows.map((row) => ({
orderItemId: Number(row.id),
canonicalProductId: null,
expectedCanonicalProductId: Number(row.expected_product_id),
})),
})
resetOrderItemCount += reset.length
if (candidates.rows.length < 20000) break
}
const deleted = await query(
`
@@ -6,6 +6,7 @@ import { resolveFinanceDateRange } from '../../utils/finance-date-range.js'
import { nowIso } from '../../utils/time.js'
import { rebuildFactsForRange } from './affiliate-product-service.js'
import { WORK_ORDER_TEMPLATE_COST_BASELINE } from './cost-center-shared.js'
import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
type WorkOrderSkuBackfillRow = {
work_order_item_count: number
@@ -64,12 +65,17 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
let ruleMatchedCount = 0
let ruleUpdatedCount = 0
for (let batch = 0; batch < 40; batch += 1) {
const targetRows = await query<{ id: number; canonical_product_id: number }>(
const targetRows = await query<{
id: number
canonical_product_id: number
expected_product_id: number | null
}>(
`
WITH work_order_item AS (
SELECT DISTINCT ON (wo.order_item_id)
wo.order_item_id AS id,
wo.product_rule_id
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 orders o ON o.id = oi.order_id
@@ -81,7 +87,7 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
ORDER BY wo.order_item_id, wo.id DESC
LIMIT $4
)
SELECT woi.id, p.id AS canonical_product_id
SELECT woi.id, p.id AS canonical_product_id, woi.expected_product_id
FROM work_order_item woi
JOIN work_product_rules wpr ON wpr.id = woi.product_rule_id
JOIN product_catalog p ON p.canonical_sku = 'tpl:' || wpr.rule_key
@@ -95,23 +101,16 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
lastRuleItemId = lastSelectedId
ruleMatchedCount += targetRows.rows.length
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = $1
FROM unnest($2::bigint[], $3::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS DISTINCT FROM input.canonical_product_id
`,
[
now,
targetRows.rows.map((row) => Number(row.id)),
targetRows.rows.map((row) => Number(row.canonical_product_id)),
],
)
ruleUpdatedCount += updated.rowCount || 0
const updated = await attachFulfillmentConfigurations({
mappings: targetRows.rows.map((row) => ({
orderItemId: Number(row.id),
canonicalProductId: Number(row.canonical_product_id),
expectedCanonicalProductId:
row.expected_product_id === null ? null : Number(row.expected_product_id),
})),
onlyWhenCanonicalProductDifferent: true,
})
ruleUpdatedCount += updated.length
if (targetRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break
}
@@ -249,23 +248,14 @@ export async function backfillAdminWorkOrderProductSkus(payload: Record<string,
if (lastSelectedId <= lastTailItemId) break
lastTailItemId = lastSelectedId
const updated = await query(
`
UPDATE order_items oi
SET canonical_product_id = input.canonical_product_id,
updated_at = $1
FROM unnest($2::bigint[], $3::bigint[])
AS input(id, canonical_product_id)
WHERE oi.id = input.id
AND oi.canonical_product_id IS NULL
`,
[
now,
tailRows.rows.map((row) => Number(row.id)),
tailRows.rows.map((row) => Number(row.canonical_product_id)),
],
)
tailUpdatedCount += updated.rowCount || 0
const updated = await attachFulfillmentConfigurations({
mappings: tailRows.rows.map((row) => ({
orderItemId: Number(row.id),
canonicalProductId: Number(row.canonical_product_id),
})),
onlyWhenCanonicalProductNull: true,
})
tailUpdatedCount += updated.length
if (tailRows.rows.length < WORK_ORDER_BACKFILL_BATCH_SIZE) break
}
@@ -90,17 +90,23 @@ test('人工确认必须单事务:扣款与快照同事务,且使用 with-cl
})
test('人工确认前必须自动匹配履约配置,配置保存/定时任务也要回填', async () => {
const [manualSource, configSource, jobSource, migrationSource] = await Promise.all([
readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'),
readFile(
new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url),
'utf8',
),
readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'),
readFile(new URL('../../db/migrations/106_manual_sku_key.sql', import.meta.url), 'utf8'),
])
const [manualSource, configSource, jobSource, migrationSource, reconcileMigration] =
await Promise.all([
readFile(new URL('./manual-fulfillment-service.ts', import.meta.url), 'utf8'),
readFile(
new URL('../cost-center/manual-fulfillment-config-service.ts', import.meta.url),
'utf8',
),
readFile(new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url), 'utf8'),
readFile(new URL('../../db/migrations/106_manual_sku_key.sql', import.meta.url), 'utf8'),
readFile(
new URL('../../db/migrations/135_reconcile_manual_standard_products.sql', import.meta.url),
'utf8',
),
])
assert.ok(
manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId, now)'),
manualSource.includes('matchManualFulfillmentConfigWithClient(client, orderItemId)') &&
manualSource.includes('attachFulfillmentConfigurationsWithExecutor(client'),
'确认时必须按 SKU 自动挂渠道',
)
assert.ok(
@@ -137,6 +143,18 @@ test('人工确认前必须自动匹配履约配置,配置保存/定时任务
migrationSource.includes('sku_name'),
'生成列必须保留 sku_code / sku_name 兜底',
)
const listStart = configSource.indexOf('export async function listManualSkus')
const listEnd = configSource.indexOf('export async function getManualSkuOverview')
const listBody = configSource.slice(listStart, listEnd)
assert.ok(
listBody.includes('FROM product_catalog p') && listBody.includes('productSourceFilterSql'),
'人工成本配置必须以 man: 标准商品为唯一清单来源',
)
assert.ok(
reconcileMigration.includes('tmp_manual_product_merge') &&
reconcileMigration.includes("status = 'archived'"),
'存量重复人工 SKU 必须合并到同一标准商品并归档重复项',
)
})
test('人工成本缺失由管道按配置与快照生成并收口', async () => {
@@ -17,6 +17,10 @@ import {
import { rebuildChannelSalesDailyFacts } from './sales-facts-channel-service.js'
import { syncFinanceLedgerFromFacts } from './finance-ledger-service.js'
import { scheduleVoucherConsumeForOrderItems } from '../fulfillment/voucher-consume-queue-service.js'
import {
attachFulfillmentConfigurationsWithExecutor,
transitionFulfillmentStateWithExecutor,
} from '../fulfillment/fulfillment-state-service.js'
import {
calculateManualCostFen,
normalizeManualCreditQuantity,
@@ -307,7 +311,7 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
const result = await withTransaction(async (client) => {
// 配置可能晚于订单创建:确认前按 SKU 再匹配一次(与定时 rematch 同口径),
// 保证「配置好之后的新单」不需要人工再点一次「重新匹配」。
await matchManualFulfillmentConfigWithClient(client, orderItemId, now)
await matchManualFulfillmentConfigWithClient(client, orderItemId)
const context = await client.query<ConfirmContextRow>(
`
@@ -621,19 +625,21 @@ export async function confirmManualFulfillment(payload: JsonObject, session: Adm
// 人工确认即渠道成立:写新三列,避免订单一直停在 pending。
const finalChannel = platformKey === 'on_account' ? 'on_account' : 'manual'
await client.query(
`
UPDATE order_items
SET fulfillment_channel = $3,
fulfillment_status = 'assigned',
fulfillment_source = 'manual_confirm',
fulfillment_decided_at = $2,
updated_at = $2
WHERE id = $1
AND fulfillment_status <> 'archived'
`,
[orderItemId, now, finalChannel],
)
await transitionFulfillmentStateWithExecutor(client, {
orderItemId,
patch: {
channel: finalChannel,
status: 'assigned',
source: 'manual_confirm',
decidedAt: now,
decidedBy: Number(session.userId || 0) || null,
pendingReason: null,
pendingDetail: null,
},
eventType: 'manual_fulfillment_confirmed',
actor: { type: 'finance', id: session.userId },
reason: platformKey,
})
// 人工缺成本异常立即收口,不等下一轮定时任务。
await client.query(
@@ -788,11 +794,7 @@ async function resolveCreditOwnerAdminUserId(
* 与 `rematchManualFulfillments` 使用同一套键:规范中文内部 SKU。
* 这样即使配置晚于订单创建,新单在确认时也能自动挂上渠道。
*/
async function matchManualFulfillmentConfigWithClient(
client: PoolClient,
orderItemId: number,
now: string,
) {
async function matchManualFulfillmentConfigWithClient(client: PoolClient, orderItemId: number) {
const matched = await client.query<{
id: number
canonical_product_id: number | null
@@ -823,17 +825,19 @@ async function matchManualFulfillmentConfigWithClient(
)
const match = matched.rows[0]
if (!match) return false
await client.query(
`
UPDATE order_items
SET fulfillment_config_id = $2,
canonical_product_id = COALESCE($3, canonical_product_id),
updated_at = $4
WHERE id = $1
AND fulfillment_config_id IS NULL
`,
[orderItemId, match.id, match.canonical_product_id, now],
)
await attachFulfillmentConfigurationsWithExecutor(client, {
mappings: [
{
orderItemId,
configId: Number(match.id),
...(match.canonical_product_id
? { canonicalProductId: Number(match.canonical_product_id) }
: {}),
},
],
onlyWhenConfigNull: true,
onlyWhenCanonicalProductNull: true,
})
return true
}
@@ -2,6 +2,7 @@ import { query, withTransaction } from '../../db/client.js'
import { syncCostSnapshotsForItems } from '../cost-center/cost-snapshot-service.js'
import { refreshFactsForDates } from '../finance/refresh-facts.js'
import { scheduleVoucherConsumeForOrderItems } from './voucher-consume-queue-service.js'
import { transitionFulfillmentStateWithExecutor } from './fulfillment-state-service.js'
/**
* 自动发货失败 → 人工兜底。
@@ -22,9 +23,16 @@ export const AFFILIATE_FAILURE_STATUSES = [
const AFFILIATE_SUCCESS_STATUSES = ['redeemed', 'paid'] as const
export async function runAffiliateFallback(
options: { limit?: number } = {},
options: { limit?: number; cutoffCreatedAt?: string } = {},
): Promise<{ scanned: number; switched: number }> {
const limit = Math.min(1000, Math.max(1, Math.trunc(options.limit || 200)))
const params: unknown[] = [AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit]
const cutoffCreatedAt = String(options.cutoffCreatedAt || '').trim()
let cutoffFilter = ''
if (cutoffCreatedAt && Number.isFinite(Date.parse(cutoffCreatedAt))) {
params.push(cutoffCreatedAt)
cutoffFilter = `AND o.created_at >= $${params.length}::timestamptz`
}
const candidates = await query<{
id: number
order_item_id: number
@@ -80,10 +88,11 @@ export async function runAffiliateFallback(
AND ft.executor_key = 'affiliate_dash'
AND ft.task_status = ANY($2::text[])
)
${cutoffFilter}
ORDER BY oi.id
LIMIT $3
`,
[AFFILIATE_FAILURE_STATUSES, AFFILIATE_SUCCESS_STATUSES, limit],
params,
)
const switchedIds: number[] = []
@@ -91,50 +100,51 @@ export async function runAffiliateFallback(
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),
canonical_product_id = COALESCE($3, canonical_product_id),
fulfillment_pending_reason = NULL,
fulfillment_pending_detail = NULL,
fulfillment_decided_at = NOW(),
item_snapshot_json = COALESCE(item_snapshot_json, '{}'::jsonb) || jsonb_build_object(
'internalSkuChannel', 'manual',
'internalSkuChannelSource', 'affiliate_fallback',
'internalSkuChannelReason', '自动发货失败,转人工兜底',
'isConfigured', TRUE,
'affiliateDash', NULL,
'fulfillmentRoute', jsonb_build_object(
'selectedExecutorKey', 'manual_dispatch',
'selectedRuleId', '',
'reason', '自动发货失败,转人工兜底',
'matchedRule', NULL,
'candidates', '[]'::jsonb,
'skipped', '[]'::jsonb
),
'classification', jsonb_build_object(
'channel', 'manual',
'canonicalSku', resolve_internal_sku(manual_sku_key, sku_name, sku_code)
)
),
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, row.config_product_id],
)
if ((result.rowCount || 0) === 0) return false
const transitioned = await transitionFulfillmentStateWithExecutor(client, {
orderItemId: Number(row.id),
expected: {
status: 'assigned',
channel: 'affiliate_dash',
locked: false,
},
patch: {
originalChannelIfNull: 'affiliate_dash',
channel: 'manual',
status: 'assigned',
source: 'affiliate_fallback',
switchReason: 'affiliate_failed',
switchedAt: new Date().toISOString(),
...(row.config_id ? { configIdIfNull: Number(row.config_id) } : {}),
...(row.config_product_id
? { canonicalProductIdIfNull: Number(row.config_product_id) }
: {}),
pendingReason: null,
pendingDetail: null,
decidedAt: new Date().toISOString(),
itemSnapshotPatch: {
internalSkuChannel: 'manual',
internalSkuChannelSource: 'affiliate_fallback',
internalSkuChannelReason: '自动发货失败,转人工兜底',
isConfigured: true,
affiliateDash: null,
fulfillmentRoute: {
selectedExecutorKey: 'manual_dispatch',
selectedRuleId: '',
reason: '自动发货失败,转人工兜底',
matchedRule: null,
candidates: [],
skipped: [],
},
classification: {
channel: 'manual',
canonicalSku: row.internal_sku,
},
},
},
eventType: 'affiliate_failed_fallback_to_manual',
reason: 'affiliate_failed',
})
if (!transitioned) return false
await client.query(
`
@@ -10,6 +10,7 @@ import {
findConflictingCarrier,
loadActiveCarrierEvidenceForItem,
} from './carrier-evidence-service.js'
import { attachFulfillmentConfigurations } from './fulfillment-state-service.js'
function parseObject(value: unknown): Record<string, unknown> {
if (value && typeof value === 'object' && !Array.isArray(value)) {
@@ -54,7 +55,11 @@ async function hasAffiliateTask(orderItemId: number): Promise<boolean> {
return result.rows[0]?.exists === true
}
const BLOCKED_INACTIVE_CARRIER_STATUSES = new Set(['closed', 'cancelled', 'manual_review'])
const BLOCKED_WORK_ORDER_STATUSES = new Set(['closed', 'cancelled', 'manual_review'])
const HARD_BLOCKED_INACTIVE_REASONS = new Set([
'carrier_task_user_refund',
'carrier_task_cancelled',
])
async function findBlockingInactiveCarrier(input: {
orderItemId: number
@@ -64,16 +69,16 @@ async function findBlockingInactiveCarrier(input: {
const result =
input.channel === 'affiliate_dash'
? await query<{ status: string }>(
`SELECT task_status AS status
? await query<{ status: string; result_code: string }>(
`SELECT task_status AS status, result_code
FROM fulfillment_tasks
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'
ORDER BY id DESC
LIMIT 1`,
[input.orderItemId],
)
: await query<{ status: string }>(
`SELECT status
: await query<{ status: string; result_code: string }>(
`SELECT status, '' AS result_code
FROM work_orders
WHERE order_item_id = $1
ORDER BY id DESC
@@ -82,7 +87,14 @@ async function findBlockingInactiveCarrier(input: {
)
const status = String(result.rows[0]?.status || '')
if (!BLOCKED_INACTIVE_CARRIER_STATUSES.has(status)) return null
const resultCode = String(result.rows[0]?.result_code || '')
const shouldBlock =
input.channel === 'affiliate_dash'
? INACTIVE_CARRIER_TASK_STATUSES.includes(
status as (typeof INACTIVE_CARRIER_TASK_STATUSES)[number],
)
: BLOCKED_WORK_ORDER_STATUSES.has(status)
if (!shouldBlock) return null
if (status === 'manual_review') {
return {
@@ -91,6 +103,34 @@ async function findBlockingInactiveCarrier(input: {
message: '已有履约载体进入人工复核,需先处理后再重试',
}
}
if (status === 'retry_pending' || status === 'failed') {
return {
status,
reason: 'carrier_task_retry_pending',
message: '已有履约任务失败待重试,需由运营确认后重新创建',
}
}
if (status === 'expired') {
return {
status,
reason: 'carrier_task_expired',
message: '已有履约任务已过期,需由运营确认后重新创建',
}
}
if (resultCode === 'USER_APPLY_REFUND') {
return {
status,
reason: 'carrier_task_user_refund',
message: '用户已申请退款,不能自动重新发货',
}
}
if (resultCode === 'affiliate_dash_cancelled') {
return {
status,
reason: 'carrier_task_cancelled',
message: '上游履约订单已取消,不能自动重新发货',
}
}
return {
status,
reason: 'carrier_task_closed',
@@ -101,6 +141,14 @@ async function findBlockingInactiveCarrier(input: {
}
}
function shouldBlockInactiveCarrier(
inactiveCarrier: { reason: string } | null,
retryInactiveCarrier: boolean,
): inactiveCarrier is { status: string; reason: string; message: string } {
if (!inactiveCarrier) return false
return !retryInactiveCarrier || HARD_BLOCKED_INACTIVE_REASONS.has(inactiveCarrier.reason)
}
function toResolvedItem(item: OrderItemRow): ResolvedFulfillmentItem {
return {
...item,
@@ -127,15 +175,8 @@ export async function ensureCarrierForOrderItem(input: {
item: OrderItemRow
channel: FulfillmentChannel
retryInactiveCarrier?: boolean
allowInactiveTaskRetry?: boolean
}): Promise<EnsureCarrierResult> {
const {
order,
item,
channel,
retryInactiveCarrier = false,
allowInactiveTaskRetry = false,
} = input
const { order, item, channel, retryInactiveCarrier = false } = input
const existingCarrier = findConflictingCarrier(
await loadActiveCarrierEvidenceForItem(Number(item.id)),
channel,
@@ -160,14 +201,18 @@ export async function ensureCarrierForOrderItem(input: {
)
const row = config.rows[0]
if (row) {
await query(
`UPDATE order_items
SET fulfillment_config_id = $2,
canonical_product_id = COALESCE(canonical_product_id, $3),
updated_at = NOW()
WHERE id = $1`,
[item.id, row.id, row.canonical_product_id],
)
await attachFulfillmentConfigurations({
mappings: [
{
orderItemId: Number(item.id),
configId: Number(row.id),
...(row.canonical_product_id
? { canonicalProductId: Number(row.canonical_product_id) }
: {}),
},
],
onlyWhenCanonicalProductNull: true,
})
return { ready: true, created: false, reason: 'manual_config_ready', message: '' }
}
return {
@@ -194,7 +239,7 @@ export async function ensureCarrierForOrderItem(input: {
orderItemId: Number(item.id),
channel,
})
if (inactiveCarrier && !retryInactiveCarrier) {
if (shouldBlockInactiveCarrier(inactiveCarrier, retryInactiveCarrier)) {
return {
ready: false,
created: false,
@@ -274,7 +319,7 @@ export async function ensureCarrierForOrderItem(input: {
orderItemId: Number(item.id),
channel,
})
if (inactiveCarrier && !retryInactiveCarrier) {
if (shouldBlockInactiveCarrier(inactiveCarrier, retryInactiveCarrier)) {
return {
ready: false,
created: false,
@@ -326,7 +371,7 @@ export async function ensureCarrierForOrderItem(input: {
item_snapshot_json: snapshot,
} as OrderItemRow,
],
{ allowInactiveTaskRetry: retryInactiveCarrier || allowInactiveTaskRetry },
{ allowInactiveTaskRetry: retryInactiveCarrier },
)
}
const ready = await hasAffiliateTask(Number(item.id))
@@ -1,4 +1,4 @@
import { query } from '../../db/client.js'
import { query, withTransaction } from '../../db/client.js'
import { getOrderById } from '../../repositories/order-repo.js'
import type { OrderItemRow } from '../../types/repository/rows.js'
import { logWarn } from '../../utils/logger.js'
@@ -8,6 +8,10 @@ import {
type FulfillmentChannel,
} from './channel-types.js'
import { ensureCarrierForOrderItem } from './carrier-creation-service.js'
import {
transitionFulfillmentState,
transitionFulfillmentStateWithExecutor,
} from './fulfillment-state-service.js'
import {
planInternalSkuChannels,
type InternalSkuChannelRow,
@@ -52,6 +56,7 @@ async function loadCandidateItems(input: {
assignedMissingCarrier?: boolean
afterId?: number
includeLocked?: boolean
cutoffCreatedAt?: string
}): Promise<CarrierCandidate[]> {
const ids = (input.orderItemIds || [])
.map(Number)
@@ -65,6 +70,12 @@ async function loadCandidateItems(input: {
params.push(Math.max(0, Math.trunc(Number(input.afterId))))
idFilter = `AND oi.id > $${params.length}`
}
const cutoffCreatedAt = String(input.cutoffCreatedAt || '').trim()
let cutoffFilter = ''
if (cutoffCreatedAt && Number.isFinite(Date.parse(cutoffCreatedAt))) {
params.push(cutoffCreatedAt)
cutoffFilter = `AND oi.created_at >= $${params.length}::timestamptz`
}
const statusFilter = input.assignedMissingCarrier
? `oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel IN ('affiliate_dash', 'work_order')
@@ -76,7 +87,14 @@ async function loadCandidateItems(input: {
SELECT 1 FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
AND ft.task_status NOT IN (${INACTIVE_CARRIER_TASK_STATUSES.map((status) => `'${status}'`).join(', ')})
)`
)
AND COALESCE((
SELECT ft.result_code
FROM fulfillment_tasks ft
WHERE ft.order_item_id = oi.id AND ft.executor_key = 'affiliate_dash'
ORDER BY ft.id DESC
LIMIT 1
), '') NOT IN ('USER_APPLY_REFUND', 'affiliate_dash_cancelled')`
: `oi.fulfillment_status = 'pending'
${input.includeLocked ? '' : 'AND oi.fulfillment_locked = FALSE'}`
@@ -116,6 +134,7 @@ async function loadCandidateItems(input: {
) capability ON jsonb_array_length(capability.channels) > 0
WHERE ${statusFilter}
${idFilter}
${cutoffFilter}
ORDER BY oi.id
LIMIT $1
`,
@@ -130,24 +149,50 @@ async function markPendingWithoutCarrier(input: {
attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }>
}): Promise<void> {
const carrierIssue = input.attempts[0]?.reason || 'carrier_attempt_failed'
await query(
`
UPDATE order_items
SET fulfillment_pending_reason = 'no_carrier',
fulfillment_pending_detail = $2::jsonb,
updated_at = NOW()
WHERE id = $1 AND fulfillment_status = 'pending'
`,
[
input.itemId,
JSON.stringify({
await transitionFulfillmentState({
orderItemId: input.itemId,
expected: { status: 'pending' },
patch: {
pendingReason: 'no_carrier',
pendingDetail: {
internalSku: input.internalSku,
carrierIssue,
carrierMessage: input.attempts[0]?.message || '',
attempts: input.attempts,
}),
],
)
},
},
eventType: 'carrier_attempt_failed',
reason: carrierIssue,
})
}
async function markAssignedMissingCarrierPending(input: {
itemId: number
previousChannel: string
internalSku: string
attempts: Array<{ channel: FulfillmentChannel; reason: string; message: string }>
}): Promise<void> {
const carrierIssue = input.attempts[0]?.reason || 'carrier_attempt_failed'
await transitionFulfillmentState({
orderItemId: input.itemId,
expected: { status: 'assigned' },
patch: {
status: 'pending',
channel: null,
source: 'auto_mapping',
locked: false,
pendingReason: 'no_carrier',
pendingDetail: {
internalSku: input.internalSku,
previousChannel: input.previousChannel,
carrierIssue,
carrierMessage: input.attempts[0]?.message || '',
attempts: input.attempts,
},
},
eventType: 'assigned_carrier_missing',
reason: carrierIssue,
})
}
async function selectSafeFallback(input: {
@@ -157,39 +202,41 @@ async function selectSafeFallback(input: {
reason: string
includeAssignedMissingCarrier: boolean
}): Promise<boolean> {
const result = await query(
`
UPDATE order_items
SET fulfillment_channel = $2,
fulfillment_status = 'assigned',
fulfillment_source = 'auto_mapping',
fulfillment_reason = $3,
channel_switch_reason = $3,
channel_switched_at = NOW(),
canonical_product_id = NULL,
fulfillment_config_id = NULL,
fulfillment_pending_reason = NULL,
fulfillment_pending_detail = NULL,
updated_at = NOW()
WHERE id = $1
AND (
(fulfillment_status = 'pending' AND fulfillment_locked = FALSE)
OR (
$4::boolean
AND fulfillment_status = 'assigned'
AND fulfillment_locked = FALSE
AND fulfillment_source NOT IN ('operator_assigned', 'manual_confirm')
)
)
`,
[
input.orderItemId,
input.channel,
`主渠道 ${input.primaryChannel} 无法创建载体:${input.reason}`,
input.includeAssignedMissingCarrier,
],
)
const updated = (result.rowCount || 0) > 0
const reason = `主渠道 ${input.primaryChannel} 无法创建载体:${input.reason}`
const updated = await withTransaction(async (client) => {
const common = {
orderItemId: input.orderItemId,
patch: {
status: 'assigned' as const,
channel: input.channel,
source: 'auto_mapping',
reason,
switchReason: reason,
switchedAt: new Date().toISOString(),
canonicalProductId: null,
configId: null,
pendingReason: null,
pendingDetail: null,
},
eventType: 'fallback_carrier_selected',
reason,
}
const pending = await transitionFulfillmentStateWithExecutor(client, {
...common,
expected: { status: 'pending', locked: false },
})
if (pending) return true
if (!input.includeAssignedMissingCarrier) return false
const assigned = await transitionFulfillmentStateWithExecutor(client, {
...common,
expected: {
status: 'assigned',
locked: false,
sourceNotIn: ['operator_assigned', 'manual_confirm'],
},
})
return Boolean(assigned)
})
if (updated) {
await scheduleVoucherConsumeForOrderItems([input.orderItemId], 'order_received').catch(
() => undefined,
@@ -213,6 +260,7 @@ export async function reconcilePendingCarriers(
afterId?: number
includeLocked?: boolean
retryInactiveCarrier?: boolean
cutoffCreatedAt?: string
} = {},
): Promise<{
scanned: number
@@ -229,6 +277,7 @@ export async function reconcilePendingCarriers(
assignedMissingCarrier: input.includeAssignedMissingCarrier === true,
...(input.afterId != null ? { afterId: input.afterId } : {}),
...(input.includeLocked ? { includeLocked: true } : {}),
...(input.cutoffCreatedAt ? { cutoffCreatedAt: input.cutoffCreatedAt } : {}),
...(input.orderItemIds ? { orderItemIds: input.orderItemIds } : {}),
})
@@ -271,20 +320,19 @@ export async function reconcilePendingCarriers(
if (attempts.length === 0) {
failed += 1
if (!input.includeAssignedMissingCarrier) {
await query(
`UPDATE order_items
SET fulfillment_pending_reason = 'routing_conflict',
fulfillment_pending_detail = $2::jsonb,
updated_at = NOW()
WHERE id = $1 AND fulfillment_status = 'pending'`,
[
item.id,
JSON.stringify({
await transitionFulfillmentState({
orderItemId: Number(item.id),
expected: { status: 'pending' },
patch: {
pendingReason: 'routing_conflict',
pendingDetail: {
internalSku: candidate.internal_sku,
channels: plan.conflicts.map((channel) => channel.channel),
}),
],
)
},
},
eventType: 'carrier_routing_conflict',
reason: 'routing_conflict',
})
}
continue
}
@@ -299,7 +347,6 @@ export async function reconcilePendingCarriers(
item,
channel: attempt.channel,
retryInactiveCarrier: input.retryInactiveCarrier === true,
allowInactiveTaskRetry: true,
})
if (!carrier.ready) {
failures.push({
@@ -361,6 +408,13 @@ export async function reconcilePendingCarriers(
internalSku: candidate.internal_sku,
attempts: failures,
})
} else {
await markAssignedMissingCarrierPending({
itemId: Number(item.id),
previousChannel: assignedChannel,
internalSku: candidate.internal_sku,
attempts: failures,
})
}
}
}
File diff suppressed because it is too large Load Diff
@@ -0,0 +1,30 @@
import { query } from '../../db/client.js'
import { APP_CONFIG_KEYS } from '../../config/app-config-keys.js'
type FulfillmentAutomationConfig = {
cutoff: string
}
export async function getFulfillmentAutomationCutoff(): Promise<string> {
const result = await query<{ config_json: FulfillmentAutomationConfig | null }>(
`SELECT config_json
FROM app_config_entries
WHERE config_key = $1
LIMIT 1`,
[APP_CONFIG_KEYS.fulfillmentAutomationCutoff],
)
const cutoff = String(result.rows[0]?.config_json?.cutoff || '').trim()
if (!cutoff || !Number.isFinite(Date.parse(cutoff))) return ''
return cutoff
}
export function fulfillmentCreatedAfterSql(
params: unknown[],
cutoff: string,
alias = 'oi',
): string {
const normalized = String(cutoff || '').trim()
if (!normalized || !Number.isFinite(Date.parse(normalized))) return ''
params.push(normalized)
return `AND ${alias}.created_at >= $${params.length}::timestamptz`
}
@@ -0,0 +1,57 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const carrierCreation = readFileSync(
new URL('./carrier-creation-service.ts', import.meta.url),
'utf8',
)
const carrierReconciliation = readFileSync(
new URL('./carrier-reconciliation-service.ts', import.meta.url),
'utf8',
)
const channelAssign = readFileSync(new URL('./channel-assign-service.ts', import.meta.url), 'utf8')
const taskRepo = readFileSync(new URL('../../repositories/task-repo.ts', import.meta.url), 'utf8')
const currentAttemptMigration = readFileSync(
new URL('../../db/migrations/138_fulfillment_current_attempt.sql', import.meta.url),
'utf8',
)
test('自动发货历史非活跃任务默认阻止后台自动重开', () => {
assert.ok(carrierCreation.includes('INACTIVE_CARRIER_TASK_STATUSES.includes'))
assert.ok(carrierCreation.includes("reason: 'carrier_task_retry_pending'"))
assert.ok(carrierCreation.includes('HARD_BLOCKED_INACTIVE_REASONS'))
assert.ok(carrierCreation.includes("'carrier_task_user_refund'"))
assert.ok(!carrierReconciliation.includes('allowInactiveTaskRetry: true'))
})
test('已分配但载体失效的订单会退回 pending/no_carrier', () => {
assert.ok(carrierReconciliation.includes('markAssignedMissingCarrierPending'))
assert.ok(carrierReconciliation.includes('transitionFulfillmentState'))
assert.ok(carrierReconciliation.includes("status: 'pending'"))
assert.ok(carrierReconciliation.includes("pendingReason: 'no_carrier'"))
assert.ok(carrierReconciliation.includes("'USER_APPLY_REFUND', 'affiliate_dash_cancelled'"))
})
test('整组补载体默认不包含锁定项,也不强制重开历史任务', () => {
assert.ok(channelAssign.includes('retryInactiveCarrier?: boolean'))
assert.ok(channelAssign.includes('includeLocked?: boolean'))
assert.ok(channelAssign.includes('retryInactiveCarrier: input.retryInactiveCarrier === true'))
assert.ok(channelAssign.includes('includeLocked: input.includeLocked === true'))
assert.ok(channelAssign.includes('unlockPendingChannelGroup'))
})
test('任务列表默认只展示每个订单项单元的最新尝试', () => {
assert.ok(taskRepo.includes('currentOnly = false'))
assert.ok(taskRepo.includes('ft.is_current = TRUE'))
})
test('创建新履约尝试时自动退休旧当前尝试', () => {
assert.ok(currentAttemptMigration.includes('attempt_no INTEGER'))
assert.ok(currentAttemptMigration.includes('is_current BOOLEAN'))
assert.ok(currentAttemptMigration.includes('uq_fulfillment_tasks_current_unit'))
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'))
})
@@ -19,6 +19,7 @@ test(
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const {
assignPendingOrderItemChannels,
assignChannelByInternalSku,
deletePendingChannelGroup,
deleteChannelRules,
@@ -29,11 +30,15 @@ test(
listPendingChannelGroups,
retryPendingCarrierGroup,
switchChannelForOrderItems,
unlockPendingChannelGroup,
upsertChannelRule,
} = await import('./channel-assign-service.js')
const { deleteManualSkuClassification, saveManualFulfillmentConfig } =
await import('../cost-center/manual-fulfillment-config-service.js')
const { runAffiliateFallback } = await import('./affiliate-fallback-service.js')
const { getFulfillmentAutomationCutoff } =
await import('./fulfillment-automation-config-service.js')
const { reconcilePendingCarriers } = await import('./carrier-reconciliation-service.js')
const { retryPendingWorkOrderRouting } = await import('./pending-channel-repair-service.js')
const {
deleteInternalSkuMapping,
@@ -43,6 +48,8 @@ test(
} = await import('./internal-sku-mapping-service.js')
const { resolveOrderItemForFulfillment } = await import('./product-resolution-service.js')
const { updateOrderItemSourceSnapshot } = await import('../../repositories/order-item-repo.js')
const { listTasks } = await import('../../repositories/task-repo.js')
const { listVoucherConsumeJobs } = await import('./voucher-consume-queue-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
const shopId = `it-routing-${suffix}`
const manualSku = `集成测试人工-${suffix}`
@@ -54,6 +61,10 @@ test(
const retryWorkOrderSku = `集成测试重试接单-${suffix}`
const manualRemoveSku = `集成测试移除人工-${suffix}`
const protectedRuleSku = `集成测试强制删除-${suffix}`
const missingVoucherSku = `集成测试无商品卡券-${suffix}`
const reactivatedVoucherSku = `集成测试归档卡券-${suffix}`
const retryPendingAffiliateSku = `集成测试失败待重试-${suffix}`
const cutoffTestSku = `集成测试自动化边界-${suffix}`
const retryWorkOrderRuleKey = `it-retry-${suffix}`
const orderIds: number[] = []
const orderItemIds: number[] = []
@@ -69,8 +80,12 @@ test(
retryWorkOrderSku,
manualRemoveSku,
protectedRuleSku,
missingVoucherSku,
reactivatedVoucherSku,
retryPendingAffiliateSku,
cutoffTestSku,
]
let generatedVoucherProductId: number | null = null
const createdVoucherProductIds: number[] = []
let autoManualProductId: number | null = null
let retryWorkOrderRuleId: number | null = null
let fallbackProfileId: number | null = null
@@ -130,6 +145,40 @@ test(
assert.equal(canonicalResolution.rows[0]?.platform_only, null)
assert.equal(canonicalResolution.rows[0]?.cjk_extension, '𠀀')
assert.equal(canonicalResolution.rows[0]?.case_variant, '黑色rock长袖t恤')
assert.ok(Number.isFinite(Date.parse(await getFulfillmentAutomationCutoff())))
const cutoffNewItemId = await createPendingOrderItem(cutoffTestSku)
const cutoffOldItemId = await createPendingOrderItem(cutoffTestSku)
await query(
`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)`,
[cutoffOldItemId],
)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES ($1, 'third_party_voucher', 10, FALSE, TRUE, '{}'::jsonb, 'integration_test')`,
[cutoffTestSku],
)
const cutoffAssignment = await assignPendingOrderItemChannels({
orderItemIds: [cutoffOldItemId, cutoffNewItemId],
batchSize: 10,
maxBatches: 1,
cutoffCreatedAt: '2020-01-01T00:00:00+08:00',
})
assert.equal(cutoffAssignment.scanned, 1)
assert.equal(cutoffAssignment.assigned, 1)
const cutoffStates = await query<{ id: number; fulfillment_status: string }>(
`SELECT id, fulfillment_status
FROM order_items
WHERE id = ANY($1::bigint[])
ORDER BY id DESC`,
[[cutoffOldItemId, cutoffNewItemId]],
)
assert.equal(cutoffStates.rows[0]?.fulfillment_status, 'pending')
assert.equal(cutoffStates.rows[1]?.fulfillment_status, 'assigned')
await query(
`INSERT INTO internal_sku_channels (internal_sku, channel, priority, is_fallback, enabled, source)
@@ -137,7 +186,13 @@ test(
[` 集成测试 规范-${suffix} `],
)
const normalizedInsert = await query<{ internal_sku: string }>(
`SELECT internal_sku FROM internal_sku_channels WHERE source = 'integration_test'`,
`SELECT internal_sku
FROM internal_sku_channels
WHERE source = 'integration_test'
AND canonicalize_internal_sku(internal_sku) = canonicalize_internal_sku($1)
ORDER BY id DESC
LIMIT 1`,
[triggerSku],
)
assert.equal(normalizedInsert.rows[0]?.internal_sku, triggerSku)
@@ -251,8 +306,6 @@ test(
'third_party_voucher',
)
assert.equal(switchedItem.rows[0]?.item_snapshot_json.isConfigured, false)
generatedVoucherProductId = Number(switchedItem.rows[0]?.canonical_product_id || 0) || null
const missingAffiliateItemId = await createPendingOrderItem(missingAffiliateSku)
const affiliateFailed = await assignChannelByInternalSku({
internalSku: missingAffiliateSku,
@@ -381,6 +434,7 @@ test(
)
const retriedCarrier = await retryPendingCarrierGroup({
internalSku: retryWorkOrderSku,
retryInactiveCarrier: true,
})
assert.equal(retriedCarrier.scanned, 1)
assert.equal(retriedCarrier.assigned, 1)
@@ -421,6 +475,24 @@ test(
{ id: crossShopB, name: '测试店铺 B' },
],
)
const pendingClassification = await listPendingChannelGroups({
keyword: crossShopSku,
issueCategory: 'classification',
page: 1,
pageSize: 10,
})
assert.equal(pendingClassification.total, 1)
assert.equal(
pendingClassification.items[0]?.issue_breakdown[0]?.issue_code,
'missing_channel',
)
const pendingAfterSales = await listPendingChannelGroups({
keyword: crossShopSku,
issueCategory: 'after_sales',
page: 1,
pageSize: 10,
})
assert.equal(pendingAfterSales.total, 0)
const deletedPending = await deletePendingChannelGroup({
internalSku: crossShopSku,
@@ -609,13 +681,21 @@ test(
[retryTaskItemId],
)
const retryTaskOrderRow = retryTaskOrder.rows[0]!
const insertRetryTask = (taskNo: string, status: string) =>
const insertRetryTask = (
taskNo: string,
status: string,
attemptNo: number,
isCurrent: boolean,
) =>
query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, created_at, updated_at
) VALUES ($1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash', $10, NOW(), NOW())`,
task_status, attempt_no, is_current, created_at, updated_at
) VALUES (
$1, $2, 1, $3, $4, $5, $6, $7, $8, $9, 'affiliate_dash',
$10, $11, $12, NOW(), NOW()
)`,
[
retryTaskOrderRow.order_id,
retryTaskItemId,
@@ -627,12 +707,14 @@ test(
retryTaskOrderRow.platform_order_id,
fallbackProfileId,
status,
attemptNo,
isCurrent,
],
)
await insertRetryTask(`IT-AFF-CLOSED-${suffix}`, 'closed')
await insertRetryTask(`IT-AFF-ACTIVE-${suffix}`, 'paid')
await insertRetryTask(`IT-AFF-CLOSED-${suffix}`, 'closed', 1, false)
await insertRetryTask(`IT-AFF-ACTIVE-${suffix}`, 'paid', 2, true)
await assert.rejects(
() => insertRetryTask(`IT-AFF-DUPLICATE-${suffix}`, 'claimed'),
() => insertRetryTask(`IT-AFF-DUPLICATE-${suffix}`, 'claimed', 3, false),
/duplicate key value violates unique constraint/,
)
const retryTaskRows = await query<{ task_status: string }>(
@@ -643,14 +725,63 @@ test(
retryTaskRows.rows.map((row) => row.task_status),
['closed', 'paid'],
)
const currentTasksOnly = await listTasks({
platformOrderId: retryTaskOrderRow.platform_order_id,
currentOnly: true,
page: 1,
pageSize: 20,
})
assert.equal(currentTasksOnly.total, 1)
assert.equal(currentTasksOnly.items[0]?.task_status, 'paid')
const allTaskAttempts = await listTasks({
platformOrderId: retryTaskOrderRow.platform_order_id,
currentOnly: false,
page: 1,
pageSize: 20,
})
assert.equal(allTaskAttempts.total, 2)
await ensureVoucherRuleForSku(voucherSku)
const voucherChannel = await query<{ channel: string; source: string }>(
`SELECT channel, source FROM internal_sku_channels
assert.equal(await ensureVoucherRuleForSku(missingVoucherSku), false)
const missingVoucherRules = await query<{ count: number }>(
`SELECT count(*)::int AS count FROM internal_sku_channels
WHERE internal_sku = $1 AND channel = 'third_party_voucher'`,
[missingVoucherSku],
)
assert.equal(missingVoucherRules.rows[0]?.count, 0)
const voucherProduct = 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', '{}'::jsonb, NOW(), NOW())
RETURNING id`,
[`tpv:${voucherSku}`, `卡券展示名-${suffix}`],
)
createdVoucherProductIds.push(Number(voucherProduct.rows[0]!.id))
assert.equal(await ensureVoucherRuleForSku(voucherSku), true)
const voucherChannel = await query<{
channel: string
source: string
canonical_product_id: string | null
priority: number
is_fallback: boolean
config_json: Record<string, unknown>
}>(
`SELECT channel, source, canonical_product_id, priority, is_fallback, config_json
FROM internal_sku_channels
WHERE internal_sku = $1 ORDER BY channel`,
[voucherSku],
)
assert.deepEqual(voucherChannel.rows, [{ channel: 'third_party_voucher', source: 'voucher' }])
assert.equal(voucherChannel.rows.length, 1)
assert.equal(voucherChannel.rows[0]?.channel, 'third_party_voucher')
assert.equal(voucherChannel.rows[0]?.source, 'voucher')
assert.equal(
Number(voucherChannel.rows[0]?.canonical_product_id),
createdVoucherProductIds[0],
)
assert.equal(voucherChannel.rows[0]?.priority, 10)
assert.equal(voucherChannel.rows[0]?.is_fallback, false)
assert.equal(voucherChannel.rows[0]?.config_json.voucherSku, voucherSku)
const legacyVoucherRules = await query<{ count: number }>(
`SELECT count(*)::int AS count FROM sku_channel_rules
WHERE match_key = $1`,
@@ -692,6 +823,233 @@ test(
const deletedRules = await deleteChannelRules([savedRule.id])
assert.equal(deletedRules.deletedCount, 1)
assert.deepEqual(deletedRules.deletedIds, [savedRule.id])
const archivedVoucherProduct = 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, '卡券', '张', 'archived',
'{"deletedAt":"2026-01-01T00:00:00.000Z","deletedBy":"integration_test"}'::jsonb,
NOW(), NOW()
)
RETURNING id`,
[`tpv:${reactivatedVoucherSku}`, reactivatedVoucherSku],
)
const archivedVoucherProductId = Number(archivedVoucherProduct.rows[0]!.id)
createdVoucherProductIds.push(archivedVoucherProductId)
const reactivatedRule = await upsertChannelRule({
matchKeyType: 'internal_sku',
matchKey: reactivatedVoucherSku,
channel: 'third_party_voucher',
priority: 80,
enabled: true,
note: '集成测试重新激活归档商品',
adminUserId: null,
})
const reactivated = await query<{
status: string
attributes_json: Record<string, unknown>
canonical_product_id: string | null
priority: number
is_fallback: boolean
}>(
`SELECT product.status, product.attributes_json,
channel.canonical_product_id, channel.priority, channel.is_fallback
FROM internal_sku_channels channel
JOIN product_catalog product ON product.id = channel.canonical_product_id
WHERE channel.id = $1`,
[reactivatedRule.id],
)
assert.equal(reactivated.rows[0]?.status, 'active')
assert.equal(Number(reactivated.rows[0]?.canonical_product_id), archivedVoucherProductId)
assert.equal(reactivated.rows[0]?.priority, 10)
assert.equal(reactivated.rows[0]?.is_fallback, false)
assert.equal(reactivated.rows[0]?.attributes_json.deletedAt, undefined)
assert.equal(typeof reactivated.rows[0]?.attributes_json.reactivatedAt, 'string')
const retryPendingItemId = await createPendingOrderItem(retryPendingAffiliateSku)
await query(
`INSERT INTO internal_sku_channels (
internal_sku, channel, priority, is_fallback, enabled, config_json, source
) VALUES (
$1, 'affiliate_dash', 10, FALSE, TRUE,
jsonb_build_object('affiliateSku', $1::text), 'integration_test'
)`,
[retryPendingAffiliateSku],
)
await query(
`UPDATE order_items
SET fulfillment_status = 'assigned',
fulfillment_channel = 'affiliate_dash',
fulfillment_source = 'carrier_evidence',
updated_at = NOW()
WHERE id = $1`,
[retryPendingItemId],
)
await query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, created_at, updated_at
)
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
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
WHERE oi.id = $1`,
[retryPendingItemId, `IT-RETRY-PENDING-${suffix}`],
)
const blockedRetry = await reconcilePendingCarriers({
orderItemIds: [retryPendingItemId],
includeAssignedMissingCarrier: true,
limit: 1,
})
assert.equal(blockedRetry.affiliateTasksCreated, 0)
const retryPendingState = await query<{
fulfillment_status: string
fulfillment_channel: string | null
fulfillment_pending_reason: string | null
fulfillment_pending_detail: Record<string, unknown>
task_count: number
}>(
`SELECT
oi.fulfillment_status,
oi.fulfillment_channel,
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
WHERE oi.id = $1`,
[retryPendingItemId],
)
assert.equal(retryPendingState.rows[0]?.fulfillment_status, 'pending')
assert.equal(retryPendingState.rows[0]?.fulfillment_channel, null)
assert.equal(retryPendingState.rows[0]?.fulfillment_pending_reason, 'no_carrier')
assert.equal(
retryPendingState.rows[0]?.fulfillment_pending_detail.carrierIssue,
'carrier_task_retry_pending',
)
assert.equal(retryPendingState.rows[0]?.task_count, 1)
await query(
`UPDATE order_items SET fulfillment_locked = TRUE, updated_at = NOW() WHERE id = $1`,
[retryPendingItemId],
)
const unlockedPendingGroup = await unlockPendingChannelGroup({
internalSku: retryPendingAffiliateSku,
})
assert.equal(unlockedPendingGroup.updated, 1)
const unlockedPendingItem = await query<{ fulfillment_locked: boolean }>(
`SELECT fulfillment_locked FROM order_items WHERE id = $1`,
[retryPendingItemId],
)
assert.equal(unlockedPendingItem.rows[0]?.fulfillment_locked, false)
await query(
`UPDATE fulfillment_tasks
SET task_status = 'closed',
result_code = 'USER_APPLY_REFUND',
updated_at = NOW()
WHERE order_item_id = $1 AND executor_key = 'affiliate_dash'`,
[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,
updated_at = NOW()
WHERE id = $1`,
[retryPendingItemId],
)
await assert.rejects(
() =>
retryPendingCarrierGroup({
internalSku: retryPendingAffiliateSku,
retryInactiveCarrier: true,
issueCodes: ['user_refund'],
}),
/不允许重试/,
)
const protectedRetryState = await query<{
fulfillment_status: string
fulfillment_pending_detail: Record<string, unknown> | null
task_count: number
}>(
`SELECT
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
WHERE oi.id = $1`,
[retryPendingItemId],
)
assert.equal(protectedRetryState.rows[0]?.fulfillment_status, 'pending')
assert.equal(protectedRetryState.rows[0]?.fulfillment_pending_detail, null)
assert.equal(protectedRetryState.rows[0]?.task_count, 1)
const voucherJobPrefix = `IT-VOUCHER-LIST-${suffix}`
const voucherJobs = await query<{ id: number }>(
`INSERT INTO voucher_consume_jobs (
provider, channel, trigger_event, target_key, platform_order_id,
due_at, status, attempt_count, max_attempts, updated_at
) VALUES
('kuaishou_industry', 'affiliate_dash', 'order_received', $1, $2, NOW(), 'succeeded', 1, 8, NOW()),
('kuaishou_industry', 'manual', 'order_received', $3, $4, NOW(), 'failed', 2, 8, NOW()),
('douyin_mashida', 'manual', 'order_received', $5, $6, NOW(), 'failed', 1, 8, NOW())
RETURNING id`,
[
`${voucherJobPrefix}-KS-AFF`,
`${voucherJobPrefix}-KS-AFF-ORDER`,
`${voucherJobPrefix}-KS-MANUAL`,
`${voucherJobPrefix}-KS-MANUAL-ORDER`,
`${voucherJobPrefix}-DY-MANUAL`,
`${voucherJobPrefix}-DY-MANUAL-ORDER`,
],
)
const firstVoucherJobPage = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
page: 1,
pageSize: 2,
})
assert.equal(firstVoucherJobPage.total, 3)
assert.equal(firstVoucherJobPage.items.length, 2)
assert.equal(firstVoucherJobPage.page, 1)
assert.equal(firstVoucherJobPage.pageSize, 2)
const secondVoucherJobPage = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
page: 2,
pageSize: 2,
})
assert.equal(secondVoucherJobPage.total, 3)
assert.equal(secondVoucherJobPage.items.length, 1)
const filteredVoucherJobs = await listVoucherConsumeJobs({
keyword: voucherJobPrefix,
provider: 'kuaishou_industry',
channel: 'manual',
status: 'failed',
page: 1,
pageSize: 20,
})
assert.equal(filteredVoucherJobs.total, 1)
assert.equal(filteredVoucherJobs.items[0]?.provider, 'kuaishou_industry')
assert.equal(filteredVoucherJobs.items[0]?.channel, 'manual')
assert.equal(filteredVoucherJobs.items[0]?.status, 'failed')
const searchedVoucherJob = await listVoucherConsumeJobs({
keyword: String(voucherJobs.rows[1]?.id || ''),
status: 'failed',
page: 1,
pageSize: 20,
})
assert.ok(searchedVoucherJob.items.some((job) => job.id === voucherJobs.rows[1]?.id))
} finally {
if (orderItemIds.length > 0) {
await query(`DELETE FROM cost_exception_items WHERE order_item_id = ANY($1::bigint[])`, [
@@ -715,6 +1073,9 @@ test(
await query(`DELETE FROM orders WHERE id = ANY($1::bigint[])`, [orderIds])
}
await query(`DELETE FROM cost_exceptions WHERE scope_key LIKE $1`, [`%${suffix}%`])
await query(`DELETE FROM voucher_consume_jobs WHERE platform_order_id LIKE $1`, [
`IT-VOUCHER-LIST-${suffix}%`,
])
await query(`DELETE FROM manual_fulfillment_configs WHERE sku_key = ANY($1::text[])`, [
internalSkus,
])
@@ -731,32 +1092,40 @@ test(
WHERE provider = '91kaquan' AND shop_id = $1 AND match_key = ANY($2::text[])`,
[shopId, internalSkus],
)
if (generatedVoucherProductId) {
await query(`DELETE FROM sales_daily_facts WHERE canonical_product_id = $1`, [
generatedVoucherProductId,
const generatedProducts = await query<{ id: string }>(
`SELECT id
FROM product_catalog
WHERE canonicalize_internal_sku(substring(canonical_sku from 5)) = ANY($1::text[])`,
[internalSkus],
)
const generatedProductIds = generatedProducts.rows.map((row) => Number(row.id))
if (generatedProductIds.length > 0) {
await query(
`DELETE FROM cost_snapshots
WHERE canonical_product_id = ANY($1::bigint[])
OR cost_version_id IN (
SELECT id FROM cost_versions WHERE canonical_product_id = ANY($1::bigint[])
)`,
[generatedProductIds],
)
await query(
`DELETE FROM sales_daily_facts WHERE canonical_product_id = ANY($1::bigint[])`,
[generatedProductIds],
)
await query(
`DELETE FROM channel_product_mappings WHERE canonical_product_id = ANY($1::bigint[])`,
[generatedProductIds],
)
await query(`DELETE FROM cost_versions WHERE canonical_product_id = ANY($1::bigint[])`, [
generatedProductIds,
])
await query(`DELETE FROM cost_snapshots WHERE canonical_product_id = $1`, [
generatedVoucherProductId,
await query(
`DELETE FROM product_channel_capabilities WHERE product_id = ANY($1::bigint[])`,
[generatedProductIds],
)
await query(`DELETE FROM product_catalog WHERE id = ANY($1::bigint[])`, [
generatedProductIds,
])
await query(`DELETE FROM channel_product_mappings WHERE canonical_product_id = $1`, [
generatedVoucherProductId,
])
await query(`DELETE FROM cost_versions WHERE canonical_product_id = $1`, [
generatedVoucherProductId,
])
await query(`DELETE FROM product_catalog WHERE id = $1 AND canonical_sku = $2`, [
generatedVoucherProductId,
`tpv:${manualSku}`,
])
}
if (autoManualProductId) {
await query(`DELETE FROM sales_daily_facts WHERE canonical_product_id = $1`, [
autoManualProductId,
])
await query(`DELETE FROM cost_versions WHERE canonical_product_id = $1`, [
autoManualProductId,
])
await query(`DELETE FROM product_catalog WHERE id = $1`, [autoManualProductId])
}
if (retryWorkOrderRuleId) {
await query(`DELETE FROM work_product_match_logs WHERE rule_id = $1`, [
@@ -0,0 +1,46 @@
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',
)
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(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('履约事件未关联影子记录'))
})
@@ -0,0 +1,368 @@
/**
* 履约状态写入口集成测试(需要已执行完全部迁移的 PostgreSQL)。
*
* FULFILLMENT_STATE_TEST_DATABASE_URL=postgres://... \
* npx tsx --test src/services/fulfillment/fulfillment-state-service.integration.test.ts
*/
import assert from 'node:assert/strict'
import test from 'node:test'
const TEST_DATABASE_URL = String(process.env.FULFILLMENT_STATE_TEST_DATABASE_URL || '').trim()
test(
'履约状态事件覆盖状态写入口、遗留直写,并保持变更幂等',
{ skip: !TEST_DATABASE_URL ? '未设置 FULFILLMENT_STATE_TEST_DATABASE_URL' : false },
async () => {
process.env.DATABASE_URL = TEST_DATABASE_URL
const { closeDb, query } = await import('../../db/client.js')
const {
assignFulfillmentChannelsWithExecutor,
attachFulfillmentConfigurations,
listFulfillmentStateEvents,
runWithFulfillmentStateTransaction,
transitionFulfillmentState,
} = await import('./fulfillment-state-service.js')
const suffix = `${Date.now()}_${Math.floor(Math.random() * 1_000_000)}`
let orderId = 0
let orderItemId = 0
let manualConfigId = 0
let productId = 0
let profileId = 0
try {
const order = await query<{ id: number }>(
`INSERT INTO orders (
provider, platform, shop_id, shop_name, platform_order_id,
order_status, pay_status, total_amount, paid_at, created_at, updated_at
) VALUES (
'91kaquan', 'kuaishou-91', $1, '状态事件集成测试', $2,
'paid', 'paid', 1000, NOW(), NOW(), NOW()
)
RETURNING id`,
[`ff-state-${suffix}`, `FF-STATE-${suffix}`],
)
orderId = Number(order.rows[0]!.id)
const item = await query<{ id: number }>(
`INSERT INTO order_items (
order_id, sku_code, sku_name, quantity, spec_json,
revenue_amount_fen, revenue_basis, created_at, updated_at
) VALUES ($1, $2, $2, 1, '{}'::jsonb, 1000, 'order_total', NOW(), NOW())
RETURNING id`,
[orderId, `状态事件测试-${suffix}`],
)
orderItemId = Number(item.rows[0]!.id)
const initialShadow = await query<{
route_status: string
carrier_status: string
issue_code: string
}>(
`SELECT route_status, carrier_status, issue_code
FROM fulfillments
WHERE order_item_id = $1`,
[orderItemId],
)
assert.equal(initialShadow.rows[0]?.route_status, 'unrouted')
assert.equal(initialShadow.rows[0]?.issue_code, 'missing_channel')
const pendingDetail = {
internalSku: `状态事件测试-${suffix}`,
channel: 'manual',
carrierIssue: 'manual_config_missing',
}
const first = await transitionFulfillmentState({
orderItemId,
expected: { status: 'pending' },
patch: {
pendingReason: 'no_carrier',
pendingDetail,
},
eventType: 'test_carrier_pending',
actor: { type: 'operator', id: 42 },
reason: 'integration_test',
})
assert.equal(first?.fulfillment_pending_reason, 'no_carrier')
let events = await listFulfillmentStateEvents(orderItemId)
assert.equal(events.length, 1)
assert.equal(events[0]?.event_type, 'test_carrier_pending')
assert.equal(events[0]?.actor_type, 'operator')
assert.equal(events[0]?.actor_id, '42')
assert.equal(events[0]?.reason, 'integration_test')
assert.deepEqual(events[0]?.next_pending_detail, pendingDetail)
await transitionFulfillmentState({
orderItemId,
expected: { status: 'pending' },
patch: {
pendingReason: 'no_carrier',
pendingDetail,
},
eventType: 'test_same_state_replay',
})
events = await listFulfillmentStateEvents(orderItemId)
assert.equal(events.length, 1, '相同状态重复写入不能制造第二条事件')
const stale = await transitionFulfillmentState({
orderItemId,
expected: { status: 'assigned' },
patch: {
channel: 'manual',
status: 'assigned',
},
eventType: 'test_stale_transition',
})
assert.equal(stale, null, '期望状态不一致时必须拒绝写入')
assert.equal((await listFulfillmentStateEvents(orderItemId)).length, 1)
await query(
`UPDATE order_items
SET fulfillment_pending_detail = jsonb_build_object('legacyWrite', TRUE),
updated_at = NOW()
WHERE id = $1`,
[orderItemId],
)
events = await listFulfillmentStateEvents(orderItemId)
assert.equal(events.length, 2, '遗留直写也必须由触发器留痕')
assert.equal(events[0]?.event_type, 'fulfillment_state_changed')
assert.equal(events[0]?.actor_type, 'system')
await query(
`UPDATE order_items
SET fulfillment_pending_detail = jsonb_build_object('legacyWrite', TRUE),
updated_at = NOW()
WHERE id = $1`,
[orderItemId],
)
events = await listFulfillmentStateEvents(orderItemId)
assert.equal(events.length, 2, '无字段变化的更新不能再记事件')
await runWithFulfillmentStateTransaction(async (client) => {
const assigned = await assignFulfillmentChannelsWithExecutor(client, {
assignments: [
{
orderItemId,
channel: 'third_party_voucher',
source: 'auto_mapping',
},
],
eventType: 'test_channel_assigned',
reason: 'integration_test',
})
assert.equal(assigned.length, 1)
assert.equal(assigned[0]?.fulfillment_status, 'assigned')
assert.equal(assigned[0]?.fulfillment_channel, 'third_party_voucher')
})
events = await listFulfillmentStateEvents(orderItemId)
assert.equal(events[0]?.event_type, 'test_channel_assigned')
assert.equal(events[0]?.previous_status, 'pending')
assert.equal(events[0]?.next_status, 'assigned')
assert.equal(events[0]?.previous_channel, null)
assert.equal(events[0]?.next_channel, 'third_party_voucher')
const assignedShadow = await query<{
route_status: string
channel: string
carrier_status: string
issue_code: string
event_count: number
}>(
`SELECT
f.route_status,
f.channel,
f.carrier_status,
f.issue_code,
(
SELECT count(*)::int
FROM fulfillment_state_events event
WHERE event.fulfillment_id = f.id
) AS event_count
FROM fulfillments f
WHERE f.order_item_id = $1`,
[orderItemId],
)
assert.equal(assignedShadow.rows[0]?.route_status, 'configured')
assert.equal(assignedShadow.rows[0]?.channel, 'third_party_voucher')
assert.equal(assignedShadow.rows[0]?.carrier_status, 'not_required')
assert.equal(assignedShadow.rows[0]?.issue_code, '')
assert.equal(Number(assignedShadow.rows[0]?.event_count), 3)
const product = 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', '{}'::jsonb, NOW(), NOW())
RETURNING id`,
[`test:ff-state-${suffix}`, `状态事件商品-${suffix}`],
)
productId = Number(product.rows[0]!.id)
const config = await query<{ id: number }>(
`INSERT INTO manual_fulfillment_configs (
sku_key, canonical_product_id, platform_key, credit_quantity,
cost_mode, enabled, note, created_at, updated_at
) VALUES ($1, $2, 'nanhang', 1, 'fixed', TRUE, 'integration_test', NOW(), NOW())
RETURNING id`,
[`状态事件测试-${suffix}`, productId],
)
manualConfigId = Number(config.rows[0]!.id)
const staleConfig = await attachFulfillmentConfigurations({
mappings: [
{
orderItemId,
configId: manualConfigId,
canonicalProductId: productId,
expectedCanonicalProductId: manualConfigId,
},
],
onlyWhenCanonicalProductNull: true,
})
assert.equal(staleConfig.length, 0, '原商品值已变化时不能覆盖并发结果')
const configured = await attachFulfillmentConfigurations({
mappings: [
{
orderItemId,
configId: manualConfigId,
canonicalProductId: productId,
},
],
expected: { status: 'assigned', channel: 'third_party_voucher' },
onlyWhenConfigNull: true,
onlyWhenCanonicalProductNull: true,
})
assert.equal(configured.length, 1)
assert.equal(Number(configured[0]?.fulfillment_config_id), manualConfigId)
assert.equal(Number(configured[0]?.canonical_product_id), productId)
const profile = await query<{ id: number }>(
`INSERT INTO fulfillment_profiles (
profile_key, name, executor_key, requires_claim, auto_dispatch,
inventory_strategy, config_json, created_at, updated_at
) VALUES ($1, '状态事件任务', 'affiliate_dash', TRUE, FALSE,
'external_platform', '{}'::jsonb, NOW(), NOW())
RETURNING id`,
[`ff-state-profile-${suffix}`],
)
profileId = Number(profile.rows[0]!.id)
await query(
`INSERT INTO fulfillment_tasks (
order_id, order_item_id, unit_index, task_no, provider, platform,
shop_id, shop_name, platform_order_id, profile_id, executor_key,
task_status, attempt_no, is_current, created_at, updated_at
)
SELECT
item.order_id, item.id, 1, $2, source.provider, source.platform,
source.shop_id, source.shop_name, source.platform_order_id, $3, 'affiliate_dash',
'paid', 1, TRUE, NOW(), NOW()
FROM order_items item
JOIN orders source ON source.id = item.order_id
WHERE item.id = $1`,
[orderItemId, `FF-SHADOW-TASK-${suffix}`, profileId],
)
const attempts = await query<{
attempt_type: string
unit_index: number
attempt_no: number
source_id: number
is_current: boolean
}>(
`SELECT attempt_type, unit_index, attempt_no, source_id, is_current
FROM fulfillment_attempts
WHERE fulfillment_id = (
SELECT id FROM fulfillments WHERE order_item_id = $1
)
ORDER BY id`,
[orderItemId],
)
assert.equal(attempts.rows.length, 1)
assert.equal(attempts.rows[0]?.attempt_type, 'task')
assert.equal(attempts.rows[0]?.unit_index, 1)
assert.equal(attempts.rows[0]?.attempt_no, 1)
assert.equal(attempts.rows[0]?.is_current, true)
const shadowDiff = await query<{
fulfillment_state_diffs: number
missing_attempts: number
missing_open_issues: number
unlinked_events: number
}>(
`SELECT
COUNT(*) FILTER (
WHERE ROW(
fulfillment.route_status,
fulfillment.channel,
fulfillment.carrier_status,
fulfillment.current_attempt_type,
fulfillment.current_attempt_source_id,
fulfillment.issue_code,
fulfillment.issue_category
) IS DISTINCT FROM ROW(
CASE WHEN state.route_status = 'configured' THEN 'configured' ELSE 'unrouted' END,
state.intended_channel,
state.carrier_status,
state.current_attempt_type,
NULLIF(state.current_attempt_id, 0),
state.issue_code,
state.issue_category
)
)::int AS fulfillment_state_diffs,
(
SELECT COUNT(*)::int
FROM fulfillment_tasks task
WHERE task.order_item_id = $1
AND NOT EXISTS (
SELECT 1 FROM fulfillment_attempts attempt
WHERE attempt.attempt_type = 'task' AND attempt.source_id = task.id
)
) AS missing_attempts,
COUNT(*) FILTER (
WHERE state.issue_code <> ''
AND NOT EXISTS (
SELECT 1 FROM fulfillment_issues issue
WHERE issue.fulfillment_id = fulfillment.id
AND issue.issue_code = state.issue_code
AND issue.status <> 'resolved'
)
)::int AS missing_open_issues,
(
SELECT COUNT(*)::int
FROM fulfillment_state_events event
WHERE event.order_item_id = $1
AND event.fulfillment_id IS NULL
) AS unlinked_events
FROM v_fulfillment_item_state state
JOIN fulfillments fulfillment ON fulfillment.order_item_id = state.order_item_id
WHERE state.order_item_id = $1`,
[orderItemId],
)
assert.equal(Number(shadowDiff.rows[0]?.fulfillment_state_diffs), 0)
assert.equal(Number(shadowDiff.rows[0]?.missing_attempts), 0)
assert.equal(Number(shadowDiff.rows[0]?.missing_open_issues), 0)
assert.equal(Number(shadowDiff.rows[0]?.unlinked_events), 0)
await assert.rejects(
query(`UPDATE fulfillment_state_events SET reason = 'tampered' WHERE order_item_id = $1`, [
orderItemId,
]),
/append-only/,
)
} finally {
if (orderItemId > 0) {
await query(`DELETE FROM order_items WHERE id = $1`, [orderItemId])
}
if (manualConfigId > 0) {
await query(`DELETE FROM manual_fulfillment_configs WHERE id = $1`, [manualConfigId])
}
if (productId > 0) {
await query(`DELETE FROM product_catalog WHERE id = $1`, [productId])
}
if (profileId > 0) {
await query(`DELETE FROM fulfillment_profiles WHERE id = $1`, [profileId])
}
if (orderId > 0) {
await query(`DELETE FROM orders WHERE id = $1`, [orderId])
}
await closeDb()
}
},
)
@@ -0,0 +1,76 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
function read(relativePath: string) {
return readFileSync(new URL(relativePath, import.meta.url), 'utf8')
}
const migration = read('../../db/migrations/140_fulfillment_state_events.sql')
const stateService = read('./fulfillment-state-service.ts')
const channelAssign = read('./channel-assign-service.ts')
const carrierReconciliation = read('./carrier-reconciliation-service.ts')
const affiliateFallback = read('./affiliate-fallback-service.ts')
const carrierCreation = read('./carrier-creation-service.ts')
const manualFulfillment = read('../finance/manual-fulfillment-service.ts')
const manualConfig = read('../cost-center/manual-fulfillment-config-service.ts')
const productNormalize = read('../cost-center/product-normalize-service.ts')
const productCatalog = read('../cost-center/product-catalog-service.ts')
const affiliateProduct = read('../cost-center/affiliate-product-service.ts')
const workOrderProduct = read('../cost-center/work-order-product-service.ts')
const templateBootstrap = read('../cost-center/template-bootstrap-service.ts')
const schedulerJobs = read('../scheduler/finance-pipeline-jobs.ts')
const taskRepo = read('../../repositories/task-repo.ts')
test('履约状态事件表记录前后状态并禁止应用修改', () => {
assert.ok(migration.includes('CREATE TABLE IF NOT EXISTS fulfillment_state_events'))
assert.ok(migration.includes('previous_pending_detail JSONB'))
assert.ok(migration.includes('next_pending_detail JSONB'))
assert.ok(migration.includes('record_fulfillment_state_event'))
assert.ok(migration.includes("current_setting('app.fulfillment_actor_type', TRUE)"))
assert.ok(migration.includes('BEFORE UPDATE ON fulfillment_state_events'))
assert.ok(migration.includes('fulfillment_state_events is append-only'))
})
test('状态写入口串行锁订单项,并在写入前校验期望状态', () => {
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'))
assert.ok(stateService.includes("set_config('app.fulfillment_event_type'"))
assert.ok(stateService.includes('itemSnapshotPatch'))
})
test('高风险履约流程统一通过状态服务写入', () => {
for (const source of [channelAssign, carrierReconciliation, affiliateFallback]) {
assert.ok(source.includes('fulfillment-state-service'))
assert.ok(!source.includes('SET fulfillment_status ='))
assert.ok(!source.includes('fulfillment_status = $'))
}
})
test('履约配置和标准商品关联统一通过状态服务的映射接口写入', () => {
for (const source of [
carrierCreation,
manualFulfillment,
manualConfig,
productNormalize,
productCatalog,
affiliateProduct,
workOrderProduct,
templateBootstrap,
schedulerJobs,
]) {
assert.ok(source.includes('fulfillment-state-service'))
assert.doesNotMatch(source, /UPDATE\s+order_items\b[\s\S]{0,300}?\bfulfillment_config_id\s*=/i)
assert.doesNotMatch(source, /UPDATE\s+order_items\b[\s\S]{0,300}?\bcanonical_product_id\s*=/i)
}
assert.ok(stateService.includes('attachFulfillmentConfigurationsWithExecutor'))
assert.ok(stateService.includes('onlyWhenCanonicalProductNull'))
})
test('任务创建先锁订单项并复用已有活跃尝试', () => {
assert.ok(taskRepo.includes('SELECT id FROM order_items WHERE id = $1 FOR UPDATE'))
assert.ok(taskRepo.includes('task_status NOT IN ('))
assert.ok(taskRepo.includes('ORDER BY is_current DESC, id DESC'))
})
@@ -0,0 +1,587 @@
import type { PoolClient, QueryResult, QueryResultRow } from 'pg'
import { query, withTransaction } from '../../db/client.js'
import type { JsonObject } from '../../types/json.js'
import type { OrderItemRow } from '../../types/repository/rows.js'
import {
isFulfillmentChannel,
type FulfillmentChannel,
type FulfillmentSource,
type FulfillmentStatus,
} from './channel-types.js'
export type FulfillmentStateExecutor = {
query: <T extends QueryResultRow = QueryResultRow>(
text: string,
params?: unknown[],
) => Promise<QueryResult<T>>
}
export type FulfillmentStateActor = {
type?: 'system' | 'operator' | 'worker' | 'finance' | 'api'
id?: number | string | null
}
export type FulfillmentStatePatch = {
status?: FulfillmentStatus
channel?: FulfillmentChannel | null
source?: FulfillmentSource | string
locked?: boolean
pendingReason?: string | null
pendingDetail?: JsonObject | null
pendingDetailUnlessCarrierIssue?: JsonObject
reason?: string | null
decidedBy?: number | null
decidedAt?: string | null
originalChannelIfNull?: FulfillmentChannel
switchReason?: string | null
switchedAt?: string | null
configId?: number | null
configIdIfNull?: number
canonicalProductId?: number | null
canonicalProductIdIfNull?: number
itemSnapshotJson?: JsonObject
itemSnapshotPatch?: JsonObject
}
export type FulfillmentStateExpected = {
status?: FulfillmentStatus | null
channel?: FulfillmentChannel | null
source?: string | null
sourceNotIn?: string[]
locked?: boolean | null
}
export type FulfillmentStateTransition = {
orderItemId: number
patch: FulfillmentStatePatch
expected?: FulfillmentStateExpected
eventType: string
actor?: FulfillmentStateActor
reason?: string
}
export type FulfillmentStateEventRow = {
[column: string]: unknown
id: number
order_item_id: number
order_id: number
event_type: string
actor_type: string
actor_id: string
reason: string
previous_status: string | null
next_status: string | null
previous_channel: string | null
next_channel: string | null
previous_source: string | null
next_source: string | null
previous_locked: boolean | null
next_locked: boolean | null
previous_pending_reason: string | null
next_pending_reason: string | null
previous_pending_detail: JsonObject | null
next_pending_detail: JsonObject | null
payload_json: JsonObject
created_at: string
}
export type FulfillmentConfigurationMapping = {
orderItemId: number
configId?: number | null
canonicalProductId?: number | null
expectedCanonicalProductId?: number | null
}
type StateColumn = {
column: string
value: unknown
cast?: 'jsonb'
mergeSnapshot?: boolean
coalesceCurrent?: boolean
preserveCarrierIssue?: boolean
}
function normalizeId(value: unknown): number {
const id = Number(value)
if (!Number.isInteger(id) || id <= 0) {
throw new TypeError('orderItemId must be a positive integer')
}
return id
}
function buildPatchColumns(patch: FulfillmentStatePatch): StateColumn[] {
if (!patch || typeof patch !== 'object') {
throw new TypeError('fulfillment state patch is required')
}
const columns: StateColumn[] = []
if (patch.status !== undefined) {
if (!['pending', 'assigned', 'archived'].includes(patch.status)) {
throw new TypeError(`invalid fulfillment status: ${patch.status}`)
}
columns.push({ column: 'fulfillment_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 })
}
if (patch.source !== undefined) {
columns.push({ column: 'fulfillment_source', value: patch.source })
}
if (patch.locked !== undefined) {
columns.push({ column: 'fulfillment_locked', value: Boolean(patch.locked) })
}
if (patch.pendingReason !== undefined) {
columns.push({ column: 'fulfillment_pending_reason', value: patch.pendingReason || null })
}
if (patch.pendingDetail !== undefined) {
columns.push({
column: 'fulfillment_pending_detail',
value: patch.pendingDetail,
cast: 'jsonb',
})
}
if (patch.pendingDetailUnlessCarrierIssue !== undefined) {
columns.push({
column: 'fulfillment_pending_detail',
value: patch.pendingDetailUnlessCarrierIssue,
cast: 'jsonb',
preserveCarrierIssue: true,
})
}
if (patch.reason !== undefined) {
columns.push({ column: 'fulfillment_reason', value: patch.reason || null })
}
if (patch.decidedBy !== undefined) {
columns.push({ column: 'fulfillment_decided_by', value: patch.decidedBy })
}
if (patch.decidedAt !== undefined) {
columns.push({ column: 'fulfillment_decided_at', value: patch.decidedAt })
}
if (patch.originalChannelIfNull !== undefined) {
if (!isFulfillmentChannel(patch.originalChannelIfNull)) {
throw new TypeError(`invalid original fulfillment channel: ${patch.originalChannelIfNull}`)
}
columns.push({
column: 'original_channel',
value: patch.originalChannelIfNull,
coalesceCurrent: true,
})
}
if (patch.switchReason !== undefined) {
columns.push({ column: 'channel_switch_reason', value: patch.switchReason || null })
}
if (patch.switchedAt !== undefined) {
columns.push({ column: 'channel_switched_at', value: patch.switchedAt })
}
if (patch.configId !== undefined) {
columns.push({ column: 'fulfillment_config_id', value: patch.configId })
}
if (patch.configIdIfNull !== undefined) {
columns.push({
column: 'fulfillment_config_id',
value: patch.configIdIfNull,
coalesceCurrent: true,
})
}
if (patch.canonicalProductId !== undefined) {
columns.push({ column: 'canonical_product_id', value: patch.canonicalProductId })
}
if (patch.canonicalProductIdIfNull !== undefined) {
columns.push({
column: 'canonical_product_id',
value: patch.canonicalProductIdIfNull,
coalesceCurrent: true,
})
}
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) {
throw new TypeError('fulfillment state patch must change at least one column')
}
return columns
}
async function setEventContext(
executor: FulfillmentStateExecutor,
transition: Pick<FulfillmentStateTransition, 'eventType'> & {
actor?: FulfillmentStateActor
reason?: string
},
) {
const eventType = String(transition.eventType || '').trim()
if (!eventType) throw new TypeError('fulfillment state eventType is required')
await executor.query(
`SELECT
set_config('app.fulfillment_event_type', $1, TRUE),
set_config('app.fulfillment_actor_type', $2, TRUE),
set_config('app.fulfillment_actor_id', $3, TRUE),
set_config('app.fulfillment_reason', $4, TRUE)`,
[
eventType,
transition.actor?.type || 'system',
String(transition.actor?.id ?? ''),
String(transition.reason || '').slice(0, 500),
],
)
}
async function clearEventContext(executor: FulfillmentStateExecutor) {
await executor
.query(
`SELECT
set_config('app.fulfillment_event_type', '', TRUE),
set_config('app.fulfillment_actor_type', '', TRUE),
set_config('app.fulfillment_actor_id', '', TRUE),
set_config('app.fulfillment_reason', '', TRUE)`,
)
.catch(() => undefined)
}
export async function transitionFulfillmentStateWithExecutor(
executor: FulfillmentStateExecutor,
input: FulfillmentStateTransition,
): Promise<OrderItemRow | null> {
const orderItemId = normalizeId(input.orderItemId)
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`,
[orderItemId],
)
const current = lockResult.rows[0]
if (!current) return null
if (
input.expected?.status !== undefined &&
current.fulfillment_status !== input.expected.status
) {
return null
}
if (
input.expected?.channel !== undefined &&
(current.fulfillment_channel ?? null) !== input.expected.channel
) {
return null
}
if (
input.expected?.source !== undefined &&
(current.fulfillment_source || '') !== (input.expected.source || '')
) {
return null
}
if (
input.expected?.sourceNotIn &&
input.expected.sourceNotIn.includes(String(current.fulfillment_source || ''))
) {
return null
}
if (
input.expected?.locked !== undefined &&
Boolean(current.fulfillment_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)
const updated = await executor.query<OrderItemRow>(
`UPDATE order_items
SET ${assignments.join(', ')}
WHERE id = $${params.length}
RETURNING *`,
params,
)
return updated.rows[0] || null
} finally {
await clearEventContext(executor)
}
}
export async function transitionFulfillmentState(
input: FulfillmentStateTransition,
): Promise<OrderItemRow | null> {
return withTransaction((client) =>
transitionFulfillmentStateWithExecutor(client as FulfillmentStateExecutor, input),
)
}
export async function transitionFulfillmentStateBatchWithExecutor(
executor: FulfillmentStateExecutor,
input: {
orderItemIds: number[]
patch: FulfillmentStatePatch
expected?: FulfillmentStateExpected
eventType: string
actor?: FulfillmentStateActor
reason?: string
},
): Promise<OrderItemRow[]> {
const ids = [
...new Set(
(input.orderItemIds || []).map(Number).filter((id) => Number.isInteger(id) && id > 0),
),
]
const results: OrderItemRow[] = []
for (const orderItemId of ids) {
const updated = await transitionFulfillmentStateWithExecutor(executor, {
orderItemId,
patch: input.patch,
eventType: input.eventType,
...(input.expected ? { expected: input.expected } : {}),
...(input.actor ? { actor: input.actor } : {}),
...(input.reason ? { reason: input.reason } : {}),
})
if (updated) results.push(updated)
}
return results
}
export async function transitionFulfillmentStateBatch(
input: Parameters<typeof transitionFulfillmentStateBatchWithExecutor>[1],
): Promise<OrderItemRow[]> {
return withTransaction((client) =>
transitionFulfillmentStateBatchWithExecutor(client as FulfillmentStateExecutor, input),
)
}
export async function assignFulfillmentChannelsWithExecutor(
executor: FulfillmentStateExecutor,
input: {
assignments: Array<{
orderItemId: number
channel: FulfillmentChannel
source: FulfillmentSource | string
}>
includeLocked?: boolean
eventType?: string
actor?: FulfillmentStateActor
reason?: string
},
): Promise<OrderItemRow[]> {
const assignments = (input.assignments || [])
.map((row) => ({
orderItemId: normalizeId(row.orderItemId),
channel: row.channel,
source: row.source,
}))
.filter((row) => isFulfillmentChannel(row.channel))
if (assignments.length === 0) return []
await setEventContext(executor, {
eventType: input.eventType || 'fulfillment_channel_assigned',
...(input.actor ? { actor: input.actor } : {}),
...(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,
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.*`,
[
assignments.map((row) => row.orderItemId),
assignments.map((row) => row.channel),
assignments.map((row) => row.source),
input.includeLocked === true,
],
)
return result.rows
} finally {
await clearEventContext(executor)
}
}
export async function attachFulfillmentConfigurationsWithExecutor(
executor: FulfillmentStateExecutor,
input: {
mappings: FulfillmentConfigurationMapping[]
expected?: Pick<FulfillmentStateExpected, 'status' | 'channel'>
onlyWhenConfigNull?: boolean
onlyWhenCanonicalProductNull?: boolean
onlyWhenCanonicalProductDifferent?: boolean
},
): Promise<OrderItemRow[]> {
const mappings = (input.mappings || []).map((row) => {
const setConfig = row.configId !== undefined
const setProduct = row.canonicalProductId !== undefined
const requireExpectedProduct = row.expectedCanonicalProductId !== undefined
if (!setConfig && !setProduct) {
throw new TypeError(
'fulfillment configuration mapping must set configId or canonicalProductId',
)
}
return {
orderItemId: normalizeId(row.orderItemId),
setConfig,
configId:
row.configId === null || row.configId === undefined ? null : normalizeId(row.configId),
setProduct,
canonicalProductId:
row.canonicalProductId === null || row.canonicalProductId === undefined
? null
: normalizeId(row.canonicalProductId),
requireExpectedProduct,
expectedCanonicalProductId:
row.expectedCanonicalProductId === null || row.expectedCanonicalProductId === undefined
? null
: normalizeId(row.expectedCanonicalProductId),
}
})
if (mappings.length === 0) return []
if (new Set(mappings.map((row) => row.orderItemId)).size !== mappings.length) {
throw new TypeError('fulfillment configuration mappings must contain unique orderItemId')
}
const params: unknown[] = [
mappings.map((row) => row.orderItemId),
mappings.map((row) => row.setConfig),
mappings.map((row) => row.configId),
mappings.map((row) => row.setProduct),
mappings.map((row) => row.canonicalProductId),
mappings.map((row) => row.requireExpectedProduct),
mappings.map((row) => row.expectedCanonicalProductId),
]
const where = ['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}`)
}
if (input.expected?.channel !== undefined) {
params.push(input.expected.channel)
where.push(`item.fulfillment_channel IS NOT DISTINCT FROM $${params.length}`)
}
if (input.onlyWhenConfigNull) {
where.push('item.fulfillment_config_id IS NULL')
}
if (input.onlyWhenCanonicalProductDifferent) {
where.push(
'(NOT input.set_product OR item.canonical_product_id IS DISTINCT FROM input.canonical_product_id)',
)
}
where.push(
'(NOT input.require_expected_product ' +
'OR item.canonical_product_id IS NOT DISTINCT FROM input.expected_canonical_product_id)',
)
const canonicalProductIfNull = input.onlyWhenCanonicalProductNull === true
params.push(canonicalProductIfNull)
const canonicalProductIfNullParam = params.length
params.push(new Date().toISOString())
const updatedAtParam = params.length
const result = await executor.query<OrderItemRow>(
`UPDATE order_items item
SET fulfillment_config_id = CASE
WHEN input.set_config THEN input.config_id
ELSE item.fulfillment_config_id
END,
canonical_product_id = CASE
WHEN NOT input.set_product THEN item.canonical_product_id
WHEN $${canonicalProductIfNullParam}::boolean
THEN COALESCE(item.canonical_product_id, input.canonical_product_id)
ELSE input.canonical_product_id
END,
updated_at = $${updatedAtParam}::timestamptz
FROM unnest(
$1::bigint[],
$2::boolean[],
$3::bigint[],
$4::boolean[],
$5::bigint[],
$6::boolean[],
$7::bigint[]
) AS input(
order_item_id,
set_config,
config_id,
set_product,
canonical_product_id,
require_expected_product,
expected_canonical_product_id
)
WHERE ${where.join(' AND ')}
RETURNING item.*`,
params,
)
return result.rows
}
export async function attachFulfillmentConfigurations(input: {
mappings: FulfillmentConfigurationMapping[]
expected?: Pick<FulfillmentStateExpected, 'status' | 'channel'>
onlyWhenConfigNull?: boolean
onlyWhenCanonicalProductNull?: boolean
onlyWhenCanonicalProductDifferent?: boolean
}): Promise<OrderItemRow[]> {
return withTransaction((client) =>
attachFulfillmentConfigurationsWithExecutor(client as FulfillmentStateExecutor, input),
)
}
export async function listFulfillmentStateEvents(
orderItemId: number | string,
options: { limit?: number } = {},
): Promise<FulfillmentStateEventRow[]> {
const id = normalizeId(orderItemId)
const limit = Math.min(500, Math.max(1, Math.trunc(Number(options.limit) || 100)))
const result = await query<FulfillmentStateEventRow>(
`SELECT *
FROM fulfillment_state_events
WHERE order_item_id = $1
ORDER BY id DESC
LIMIT $2`,
[id, limit],
)
return result.rows
}
export async function runWithFulfillmentStateTransaction<T>(
fn: (client: PoolClient) => Promise<T>,
): Promise<T> {
return withTransaction(fn)
}
@@ -0,0 +1,73 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const migration = readFileSync(
new URL('../../db/migrations/137_fulfillment_item_state_view.sql', import.meta.url),
'utf8',
)
const channelAssign = readFileSync(new URL('./channel-assign-service.ts', import.meta.url), 'utf8')
const auditScript = readFileSync(
new URL('../../../scripts/audit-fulfillment-state.ts', import.meta.url),
'utf8',
)
const automationCutoffMigration = readFileSync(
new URL('../../db/migrations/139_fulfillment_automation_cutoff.sql', import.meta.url),
'utf8',
)
const automationConfig = readFileSync(
new URL('./fulfillment-automation-config-service.ts', import.meta.url),
'utf8',
)
const schedulerJobs = readFileSync(
new URL('../scheduler/finance-pipeline-jobs.ts', import.meta.url),
'utf8',
)
test('履约读模型按订单项拆分渠道、载体、尝试、异常和锁定原因', () => {
assert.ok(migration.includes('CREATE OR REPLACE VIEW v_fulfillment_item_state'))
assert.ok(migration.includes('current_attempt_id'))
assert.ok(migration.includes('carrier_status'))
assert.ok(migration.includes('issue_code'))
assert.ok(migration.includes("'after_sales'"))
assert.ok(migration.includes("'manual_review'"))
assert.ok(migration.includes("fulfillment_status IN ('pending', 'assigned')"))
})
test('待分类查询使用统一读模型并返回异常数量', () => {
assert.ok(channelAssign.includes('FROM v_fulfillment_item_state v'))
assert.ok(channelAssign.includes('issue_breakdown'))
assert.ok(channelAssign.includes('retryable_unlocked_item_count'))
assert.ok(channelAssign.includes('auto_retryable_unlocked_item_count'))
assert.ok(channelAssign.includes('PENDING_ISSUE_CATEGORIES'))
assert.ok(channelAssign.includes('issueCategoryInput'))
})
test('重试必须按可重试 issueCode 限定,不能按整个 SKU 处理售后和复核项', () => {
assert.ok(channelAssign.includes('RETRYABLE_CARRIER_ISSUES'))
assert.ok(channelAssign.includes('issueCodes?: string[]'))
assert.ok(channelAssign.includes('oi.issue_code = ANY($2::text[])'))
})
test('保存渠道、解除锁定、归档和重试都按当前问题分类限定作用域', () => {
assert.ok(channelAssign.includes('issueCategory?: string'))
assert.ok(channelAssign.includes('state.issue_category = $'))
assert.ok(channelAssign.includes('issueCodes?: string[]'))
assert.ok(channelAssign.includes('orderItemIds?: number[]'))
})
test('履约审计脚本覆盖状态不一致、售后队列和重复尝试', () => {
assert.ok(auditScript.includes('已分配但无有效载体'))
assert.ok(auditScript.includes('售后关闭仍进入履约队列'))
assert.ok(auditScript.includes('近 24 小时重复尝试订单项'))
assert.ok(auditScript.includes('--strict'))
})
test('历史履约数据不进入自动任务,只处理自动化生效时间之后的新订单', () => {
assert.ok(automationCutoffMigration.includes('fulfillment_automation_cutoff'))
assert.ok(automationConfig.includes('getFulfillmentAutomationCutoff'))
assert.ok(automationConfig.includes('fulfillmentCreatedAfterSql'))
assert.ok(schedulerJobs.includes('cutoffCreatedAt'))
assert.ok(schedulerJobs.includes('getFulfillmentAutomationCutoff'))
assert.ok(schedulerJobs.includes('runAffiliateFallback'))
})
@@ -0,0 +1,24 @@
import assert from 'node:assert/strict'
import { readFileSync } from 'node:fs'
import test from 'node:test'
const source = readFileSync(new URL('./voucher-consume-queue-service.ts', import.meta.url), 'utf8')
test('核销任务列表支持关键词、平台、渠道和状态组合筛选', () => {
assert.ok(source.includes('keyword?: string'))
assert.ok(source.includes('provider?: string'))
assert.ok(source.includes('channel?: string'))
assert.ok(source.includes('status?: string'))
assert.ok(source.includes('provider = $'))
assert.ok(source.includes('channel = $'))
assert.ok(source.includes('status = $'))
assert.ok(source.includes('platform_order_id ILIKE'))
assert.ok(source.includes('target_key ILIKE'))
})
test('核销任务列表使用服务端总数和分页参数', () => {
assert.ok(source.includes('SELECT COUNT(*) AS total FROM voucher_consume_jobs'))
assert.ok(source.includes('LIMIT $'))
assert.ok(source.includes('OFFSET $'))
assert.ok(source.includes('ORDER BY updated_at DESC, id DESC'))
})
@@ -22,6 +22,8 @@ import { attachKuaishouIndustryVoucherToTask } from '../platforms/kuaishou-indus
import { verifyCertificate } from '../redeem/certificate-service.js'
import { logIntegration, logWarn } from '../../utils/logger.js'
import {
VOUCHER_CONSUME_CHANNELS,
VOUCHER_CONSUME_PROVIDERS,
getVoucherConsumePolicy,
resolveVoucherConsumePolicy,
type VoucherConsumeChannel,
@@ -99,6 +101,13 @@ type VoucherOrderItemTargetRow = {
const RETRY_DELAYS_SECONDS = [10, 30, 60, 300, 600, 900, 1800]
const PROCESSING_LOCK_SECONDS = 120
const VOUCHER_CONSUME_JOB_STATUSES = [
'pending',
'processing',
'succeeded',
'failed',
'cancelled',
] as const
export async function scheduleVoucherConsumeForTask(
task: TaskRow,
@@ -443,6 +452,9 @@ export async function runVoucherConsumeDispatchJob(
export async function listVoucherConsumeJobs(
input: {
keyword?: string
provider?: string
channel?: string
status?: string
page?: number
pageSize?: number
@@ -450,10 +462,43 @@ export async function listVoucherConsumeJobs(
) {
const page = clampInteger(input.page, 1, 1, Number.MAX_SAFE_INTEGER)
const pageSize = clampInteger(input.pageSize, 20, 1, 100)
const status = String(input.status || '').trim()
const keyword = String(input.keyword || '')
.trim()
.slice(0, 100)
const providerInput = String(input.provider || '').trim()
const provider = (VOUCHER_CONSUME_PROVIDERS as readonly string[]).includes(providerInput)
? providerInput
: ''
const channel = (VOUCHER_CONSUME_CHANNELS as readonly string[]).includes(
String(input.channel || '').trim(),
)
? String(input.channel).trim()
: ''
const status = (VOUCHER_CONSUME_JOB_STATUSES as readonly string[]).includes(
String(input.status || '').trim(),
)
? String(input.status).trim()
: ''
const params: unknown[] = []
const where: string[] = []
if (keyword) {
params.push(`%${keyword}%`)
where.push(
`(CAST(id AS TEXT) ILIKE $${params.length}
OR platform_order_id ILIKE $${params.length}
OR target_key ILIKE $${params.length}
OR CAST(order_id AS TEXT) ILIKE $${params.length})`,
)
}
if (provider) {
params.push(provider)
where.push(`provider = $${params.length}`)
}
if (channel) {
params.push(channel)
where.push(`channel = $${params.length}`)
}
if (status) {
params.push(status)
where.push(`status = $${params.length}`)
@@ -66,7 +66,7 @@ function stubCreatedTask(id: number, input: CreatedTaskInput): TaskRow {
unit_index: Number(input.unitIndex || 1),
platform_order_id: '',
profile_id: 0,
task_no: 'DT-test',
task_no: String(input.taskNo || 'DT-test'),
executor_key: String(input.executorKey || ''),
task_status: String(input.taskStatus || 'pending'),
delivery_status: 'pending',
@@ -428,6 +428,51 @@ test('syncDeliveryTasksForOrder 并发创建冲突时复用已落库任务', asy
assert.equal(eventCount, 0)
})
test('createTask 幂等返回已有活跃任务时不重复记计划事件', async () => {
const existingTask = {
...stubCreatedTask(402, {
orderId: 61,
orderItemId: 51,
unitIndex: 1,
executorKey: 'affiliate_dash',
}),
task_no: 'DT-existing',
}
let eventCount = 0
const tasks = await syncDeliveryTasksForOrderWithDeps(
stubOrder({ id: 61, platform_order_id: 'idempotent-create-order' }),
[
stubOrderItem({
id: 51,
order_id: 61,
sku_code: 'SKU-idempotent',
item_snapshot_json: buildAffiliateSnapshot('AFF-IDEMPOTENT', '商品 IDEMPOTENT'),
}),
],
{
loadActiveCarrierEvidence: async () => new Map(),
listTasksByOrderId: async () => [],
getFulfillmentProfileByKey: async () => buildAffiliateProfile(),
createTask: async () => existingTask,
createTaskEvent: async () => {
eventCount += 1
return null
},
logInfo: () => null,
logWarn: () => null,
nowIso: () => '2026-05-29T09:05:46.658Z',
randomId: () => 'DT-reused',
},
)
assert.deepEqual(
tasks.map((task) => task.id),
[402],
)
assert.equal(eventCount, 0)
})
test('来源重推时已分配其它渠道的订单不会补建 affiliate 任务', async () => {
let createdCount = 0
const tasks = await syncDeliveryTasksForOrderWithDeps(
@@ -181,8 +181,9 @@ export async function syncDeliveryTasksForOrderWithDeps(
? resolvePaidTaskStatus(plan.profile)
: TASK_STATUS.PENDING_PAYMENT
let createdFresh = true
let createdFresh = false
let task: TaskRow | null
const plannedTaskNo = createRandomId('DT')
try {
task = await createDeliveryTask({
orderId: order.id,
@@ -193,7 +194,7 @@ export async function syncDeliveryTasksForOrderWithDeps(
shopId: order.shop_id,
shopName: order.shop_name,
platformOrderId: order.platform_order_id,
taskNo: createRandomId('DT'),
taskNo: plannedTaskNo,
profileId: plan.profileId,
executorKey: plan.executorKey,
taskStatus: initialStatus,
@@ -211,6 +212,7 @@ export async function syncDeliveryTasksForOrderWithDeps(
createdAt,
updatedAt: createdAt,
})
createdFresh = task?.task_no === plannedTaskNo
} catch (error) {
const concurrentTasks = await listTasks(order.id)
task =
@@ -229,6 +231,14 @@ export async function syncDeliveryTasksForOrderWithDeps(
profileId: plan.profileId,
profileKey: plan.profileKey,
executorKey: plan.executorKey,
candidateTasks: concurrentTasks
.filter((candidate) => candidate.order_item_id === item.id)
.map((candidate) => ({
id: candidate.id,
unitIndex: candidate.unit_index,
status: candidate.task_status,
isCurrent: candidate.is_current,
})),
error: error instanceof Error ? error.message : String(error),
})
throw error
@@ -3,6 +3,11 @@ import { query } from '../../db/client.js'
import { assignPendingOrderItemChannels } from '../fulfillment/channel-assign-service.js'
import { runAffiliateFallback } from '../fulfillment/affiliate-fallback-service.js'
import { reconcilePendingCarriers } from '../fulfillment/carrier-reconciliation-service.js'
import { attachFulfillmentConfigurations } from '../fulfillment/fulfillment-state-service.js'
import {
fulfillmentCreatedAfterSql,
getFulfillmentAutomationCutoff,
} from '../fulfillment/fulfillment-automation-config-service.js'
import { processDueOrderServiceAutoPublishJobs } from '../worker-platform/order-service-auto-publish-service.js'
import {
resolveExceptionsForItemsByReason,
@@ -52,12 +57,14 @@ function clampInt(value: unknown, fallback: number, min: number, max: number): n
export async function runChannelAssignJob(job: JsonObject = {}) {
const config = jobConfig(job)
const cutoffCreatedAt = await getFulfillmentAutomationCutoff()
const result = await assignPendingOrderItemChannels({
batchSize: clampInt(config.batchSize, 500, 1, 5000),
maxBatches: clampInt(config.maxBatches, 20, 1, 200),
...(config.afterId != null
? { afterId: clampInt(config.afterId, 0, 0, Number.MAX_SAFE_INTEGER) }
: {}),
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
})
await touchWatermark('channel_assign', { lastItemId: result.nextCursor })
@@ -85,6 +92,7 @@ export async function runChannelAssignJob(job: JsonObject = {}) {
export async function runCarrierDispatchJob(job: JsonObject = {}) {
const config = jobConfig(job)
const cutoffCreatedAt = await getFulfillmentAutomationCutoff()
const limit = clampInt(config.limit, 200, 1, 2000)
const reconcileCursor =
config.carrierReconcileAfterId != null
@@ -93,13 +101,19 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
const reconciled = await reconcilePendingCarriers({
limit: Math.min(limit, 200),
afterId: reconcileCursor,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
})
await touchWatermark('carrier_reconcile', { lastItemId: reconciled.nextCursor })
const repairedAssigned = await reconcilePendingCarriers({
limit: Math.min(limit, 200),
includeAssignedMissingCarrier: true,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
})
await assignPendingOrderItemChannels({ batchSize: 500, maxBatches: 1 }).catch(() => undefined)
await assignPendingOrderItemChannels({
batchSize: 500,
maxBatches: 1,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
}).catch(() => undefined)
const cursor =
config.afterId != null
? Math.max(0, Math.trunc(Number(config.afterId)))
@@ -109,6 +123,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
'SELECT canonicalize_internal_sku(sku_key) AS sku_key FROM manual_fulfillment_configs WHERE enabled',
)
const configKeys = configKeysResult.rows.map((row) => row.sku_key)
const manualParams: unknown[] = [limit, cursor, configKeys]
const manualCutoff = fulfillmentCreatedAfterSql(manualParams, cutoffCreatedAt)
const manual = await query<{ id: number; sku_key: string; config_id: number }>(
`
@@ -126,26 +142,29 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND oi.fulfillment_config_id IS NULL
AND resolve_internal_sku(oi.manual_sku_key, oi.sku_name, oi.sku_code) = ANY($3::text[])
AND oi.id > $2
${manualCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor, configKeys],
manualParams,
)
let attached = 0
for (const row of manual.rows) {
const updated = await query(
`
UPDATE order_items
SET fulfillment_config_id = $2, updated_at = NOW()
WHERE id = $1 AND fulfillment_config_id IS NULL
`,
[row.id, row.config_id],
)
attached += updated.rowCount || 0
if (manual.rows.length > 0) {
const configured = await attachFulfillmentConfigurations({
mappings: manual.rows.map((row) => ({
orderItemId: Number(row.id),
configId: Number(row.config_id),
})),
expected: { status: 'assigned', channel: 'manual' },
onlyWhenConfigNull: true,
})
attached = configured.length
}
// 人工渠道但没有配置:进「人工 SKU 未配置」队列。
const missingManualParams: unknown[] = [limit, cursor, configKeys]
const missingManualCutoff = fulfillmentCreatedAfterSql(missingManualParams, cutoffCreatedAt)
const missingManual = await query<{ id: number; sku_key: string }>(
`
SELECT oi.id,
@@ -164,10 +183,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ce.status <> 'resolved'
)
AND oi.id > $2
${missingManualCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor, configKeys],
missingManualParams,
)
const exceptionEntries: AggregatedExceptionEntry[] = []
@@ -181,6 +201,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 已配置人工 SKU、但还没确认履约的单:进入「人工成本未记录」队列。
const pendingManualCostParams: unknown[] = [limit, cursor]
const pendingManualCostCutoff = fulfillmentCreatedAfterSql(
pendingManualCostParams,
cutoffCreatedAt,
)
const pendingManualCost = await query<{ id: number; sku_key: string }>(
`
SELECT oi.id,
@@ -204,10 +229,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ce.status <> 'resolved'
)
AND oi.id > $2
${pendingManualCostCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor],
pendingManualCostParams,
)
for (const row of pendingManualCost.rows) {
@@ -220,6 +246,8 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 自动发货 / 接单平台被运营指定后却没有载体:不能静默卡住,进问题队列。
const missingCarrierParams: unknown[] = [limit, cursor]
const missingCarrierCutoff = fulfillmentCreatedAfterSql(missingCarrierParams, cutoffCreatedAt)
const missingCarrierRaw = await query<{
id: number
channel: string
@@ -241,10 +269,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
)
AND oi.id > $2
${missingCarrierCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit, cursor],
missingCarrierParams,
)
const queuedCarrier = await query<{ order_item_id: number }>(
@@ -273,6 +302,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
}
// 已分配订单出现两个真实执行方时不能继续算正常渠道,必须进冲突队列人工处置。
const conflictingCarrierParams: unknown[] = [limit]
const conflictingCarrierCutoff = fulfillmentCreatedAfterSql(
conflictingCarrierParams,
cutoffCreatedAt,
)
const conflictingCarriers = await query<{
id: number
channel: string
@@ -299,10 +333,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
)
)::int > 1
${conflictingCarrierCutoff}
ORDER BY oi.id
LIMIT $1
`,
[limit],
conflictingCarrierParams,
)
for (const row of conflictingCarriers.rows) {
exceptionEntries.push({
@@ -331,8 +366,11 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
maxId = Math.max(maxId, Number(row.id))
}
const resolved = await resolveCarrierDispatchExceptions()
const fallback = await runAffiliateFallback({ limit: 200 }).catch(() => ({
const resolved = await resolveCarrierDispatchExceptions(cutoffCreatedAt)
const fallback = await runAffiliateFallback({
limit: 200,
...(cutoffCreatedAt ? { cutoffCreatedAt } : {}),
}).catch(() => ({
scanned: 0,
switched: 0,
}))
@@ -360,17 +398,24 @@ export async function runCarrierDispatchJob(job: JsonObject = {}) {
/**
* 载体/配置补上后,把对应异常自动关掉(每轮最多处理 500 条)。
*/
async function resolveCarrierDispatchExceptions(): Promise<number> {
async function resolveCarrierDispatchExceptions(cutoffCreatedAt: string): Promise<number> {
// 游标会跳过已扫过的订单项;配置是后来补的,这里兜底把配置挂上。
await query(
const attachParams: unknown[] = []
const attachCutoff = fulfillmentCreatedAfterSql(attachParams, cutoffCreatedAt)
const attachCandidates = await query<{ order_item_id: number; config_id: number }>(
`
UPDATE order_items oi
SET fulfillment_config_id = cfg.id, updated_at = NOW()
FROM manual_fulfillment_configs cfg
SELECT DISTINCT ON (oi.id)
oi.id AS order_item_id,
cfg.id AS config_id
FROM order_items 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
)
AND cfg.enabled
WHERE canonicalize_internal_sku(cfg.sku_key) = resolve_internal_sku(
oi.manual_sku_key, oi.sku_name, oi.sku_code
)
AND cfg.enabled
AND oi.fulfillment_status = 'assigned'
AND oi.fulfillment_channel = 'manual'
AND oi.fulfillment_config_id IS NULL
@@ -382,9 +427,29 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
AND ce.reason = 'manual_config_missing'
AND ce.status <> 'resolved'
)
${attachCutoff}
ORDER BY oi.id, cfg.id
LIMIT 500
`,
attachParams,
)
if (attachCandidates.rows.length > 0) {
await attachFulfillmentConfigurations({
mappings: attachCandidates.rows.map((row) => ({
orderItemId: Number(row.order_item_id),
configId: Number(row.config_id),
})),
expected: { status: 'assigned', channel: 'manual' },
onlyWhenConfigNull: true,
})
}
const carrierParams: unknown[] = []
const carrierCutoff = fulfillmentCreatedAfterSql(carrierParams, cutoffCreatedAt)
const configParams: unknown[] = []
const configCutoff = fulfillmentCreatedAfterSql(configParams, cutoffCreatedAt)
const manualCostParams: unknown[] = []
const manualCostCutoff = fulfillmentCreatedAfterSql(manualCostParams, cutoffCreatedAt)
const [carrier, config, manualCost] = await Promise.all([
query<{ order_item_id: number }>(
`
@@ -404,8 +469,10 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
AND ft.task_status NOT IN ('closed', 'expired', 'failed', 'manual_review', 'retry_pending', 'cancelled')
))
)
${carrierCutoff}
LIMIT 500
`,
carrierParams,
),
query<{ order_item_id: number }>(
`
@@ -416,14 +483,17 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
WHERE ce.reason = 'manual_config_missing'
AND ce.status <> 'resolved'
AND oi.fulfillment_config_id IS NOT NULL
${configCutoff}
LIMIT 500
`,
configParams,
),
query<{ order_item_id: number }>(
`
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
WHERE ce.reason = 'manual_cost_missing'
AND ce.status <> 'resolved'
AND EXISTS (
@@ -432,8 +502,10 @@ async function resolveCarrierDispatchExceptions(): Promise<number> {
AND cs.order_item_id = i.order_item_id
AND cs.superseded_at IS NULL AND reversed_at IS NULL
)
${manualCostCutoff}
LIMIT 500
`,
manualCostParams,
),
])
@@ -0,0 +1,12 @@
import assert from 'node:assert/strict'
import test from 'node:test'
import { normalizeInventoryDate } from './consignment-admin-service.js'
test('寄售仓库入库日期筛选只接受合法的 YYYY-MM-DD', () => {
assert.equal(normalizeInventoryDate('2026-11-05'), '2026-11-05')
assert.equal(normalizeInventoryDate(' 2026-11-05 '), '2026-11-05')
assert.equal(normalizeInventoryDate('2026-02-31'), undefined)
assert.equal(normalizeInventoryDate('2026/11/05'), undefined)
assert.equal(normalizeInventoryDate(''), undefined)
})
@@ -391,6 +391,8 @@ export async function fetchAdminConsignmentInventories(query: JsonObject = {}) {
const allowMatching =
query.allowMatching !== undefined ? String(query.allowMatching) === 'true' : undefined
const keyword = query.keyword ? String(query.keyword).trim() : undefined
const dateFrom = normalizeInventoryDate(query.dateFrom)
const dateTo = normalizeInventoryDate(query.dateTo)
const page = query.page ? Number(query.page) : 1
const pageSize = query.pageSize ? Number(query.pageSize) : 20
@@ -406,6 +408,8 @@ export async function fetchAdminConsignmentInventories(query: JsonObject = {}) {
diamondLevel,
allowMatching,
keyword,
dateFrom,
dateTo,
page,
pageSize,
})
@@ -416,6 +420,20 @@ export async function fetchAdminConsignmentInventories(query: JsonObject = {}) {
}
}
/**
* 寄售仓库的“通过审核入库时间”按自然日筛选。
* 只接受 YYYY-MM-DD,并通过 UTC roundtrip 拦截 2026-02-31 这类会被 Date 溢出的日期。
*/
export function normalizeInventoryDate(rawValue: unknown) {
const normalized = String(rawValue || '').trim()
if (!/^\d{4}-\d{2}-\d{2}$/.test(normalized)) return undefined
const parsed = new Date(`${normalized}T00:00:00.000Z`)
return Number.isNaN(parsed.getTime()) || parsed.toISOString().slice(0, 10) !== normalized
? undefined
: normalized
}
export async function fetchAdminConsignmentStats() {
return getConsignmentInventoryStats()
}