Files
baiyue-firstleg/scripts/admin-test-env-modules/grouped-dispatch-acceptance.mjs
T
2026-10-20 15:48:38 +08:00

278 lines
23 KiB
JavaScript

import { createHash } from 'node:crypto'
import { mkdir, open, writeFile, rename } from 'node:fs/promises'
import { resolve, dirname } from 'node:path'
export const DISPATCH_BASE = '/api/platform/v1/last-leg/dispatches'
export const id = value => value == null ? '' : String(value)
export const canonical = value => JSON.stringify(value)
export const success = response => response?.http === 200 && String(response.body?.code) === '200'
const positiveId = value => /^[1-9]\d*$/.test(id(value))
const amount = value => (typeof value === 'number' || typeof value === 'string' && /^\d+$/.test(value)) && Number.isSafeInteger(Number(value)) && Number(value) > 0
export class DispatchNotRun extends Error { constructor(message) { super(message); this.notRun = true } }
export function version(value) {
if (!(typeof value === 'number' && Number.isSafeInteger(value) && value >= 0 || typeof value === 'string' && /^(0|[1-9]\d*)$/.test(value))
|| BigInt(value) > 9223372036854775807n) throw Error('缺少服务器recordVersion或精度无效,不能猜测零版本')
return value
}
export const versionAfter = (next, previous) => BigInt(version(next)) > BigInt(version(previous))
// 主run提供真实原子落盘;独立调用者必须显式提供checkpoint,禁止内存意图冒充持久化。
export async function dispatchEvidenceContext(context) {
if (typeof context.checkpoint === 'function') return context
if (typeof context.marker !== 'string' || !/^[A-Za-z0-9_-]+$/.test(context.marker)) throw new DispatchNotRun('缺少可用于独占检查点的本轮marker')
const path = resolve(dirname(context.options?.output || 'reports/admin-test-env.json'), 'phase4-dispatch-checkpoints', `${context.marker}.json`)
const redact = value => {
if (Array.isArray(value)) return value.map(redact)
if (value && typeof value === 'object') return Object.fromEntries(Object.entries(value)
.filter(([name]) => !/password|authorization|secret|apiKey|credential|accessToken|refreshToken|idToken|^token$|base64|^bytes$|^content$/i.test(name))
.map(([name, nested]) => [name, redact(nested)]))
return value
}
await mkdir(dirname(path), { recursive: true })
let reservation
try { reservation = await open(path, 'wx') }
catch (error) { if (error.code === 'EEXIST') throw Error(`已存在本轮排车检查点,须核对原请求结果,禁止自动重跑:${path}`); throw error }
await reservation.close()
const scoped = Object.create(context)
scoped.checkpoint = async () => {
const resources = context.createdResources
const evidence = redact({ marker: context.marker, boundary: resources.dispatchWriteBoundary,
phase4OverseasDispatch: resources.phase4OverseasDispatch, groupedDispatch: resources.groupedDispatch })
await writeFile(`${path}.tmp`, `${JSON.stringify(evidence, null, 2)}\n`, 'utf8')
await rename(`${path}.tmp`, path)
}
context.createdResources.phase4OverseasDispatch.checkpointPath = path
await scoped.checkpoint()
return scoped
}
export function query(path, values) {
const params = new URLSearchParams()
for (const [key, value] of Object.entries(values)) {
if (value === undefined || value === null || value === '') continue
if (Array.isArray(value)) {
for (const item of value) if (item !== undefined && item !== null && item !== '') params.append(key, String(item))
continue
}
params.set(key, String(value))
}
return `${path}?${params}`
}
export function commandKey(marker, namespace, action) {
return `D4-${createHash('sha256').update(`${marker}\n${namespace}\n${action}`).digest('hex').slice(0, 48)}`
}
// 仅接受当前Candidate DTO;来源摘要、客户、仓与FBA批次均取生产回执。
export function sourceSelection(candidate) {
const source = candidate?.selection
if (!positiveId(candidate?.customerId) || !positiveId(candidate?.warehouseId) || !amount(candidate?.availablePackages)
|| !source || !/^[a-f\d]{64}$/i.test(source.sourceVersion || '') || source.sourceVersion !== candidate.sourceVersion
|| !positiveId(source.warehousePalletId) || !Array.isArray(candidate.blockReasons) || candidate.blockReasons.length
|| !Array.isArray(candidate.appointmentBatches) || !candidate.destination?.key) throw new DispatchNotRun('缺少完整、可选的本轮原生来源回执')
const expected = source.sourceType === 'INBOUND_PLAN' && positiveId(source.planPalletId) && source.returnReceiptId == null && source.relabelNo == null ? `PLAN:${source.planPalletId}`
: source.sourceType === 'RETURN_RECEIPT' && positiveId(source.returnReceiptId) && source.planPalletId == null && source.relabelNo == null ? `RETURN:${source.returnReceiptId}`
: source.sourceType === 'RELABEL' && source.relabelNo && source.planPalletId == null && source.returnReceiptId == null ? `RELABEL:${source.relabelNo}:${source.warehousePalletId}` : null
if (!expected || candidate.sourceKey !== expected) throw new DispatchNotRun('来源选择器或服务器来源键不完整')
return { sourceType: source.sourceType, planPalletId: source.planPalletId, returnReceiptId: source.returnReceiptId,
relabelNo: source.relabelNo, warehousePalletId: source.warehousePalletId, sourceVersion: source.sourceVersion }
}
export function createBody(marker, namespace, candidates) {
if (!candidates.length || new Set(candidates.map(c => id(c.warehouseId))).size !== 1) throw new DispatchNotRun('整车缺少同仓真实货物')
const cargo = candidates.map(sourceSelection), batchAllocations = []
for (const candidate of candidates) {
const batches = candidate.appointmentBatches
if (!batches.length) continue
if (batches.length !== 1 || !positiveId(candidate.selection.planPalletId)) throw new DispatchNotRun('来源跨多个FBA批次,缺少本轮明确逐批实物分配,不能推算')
const batch = batches[0]
if (!/^FBA_APPOINTMENT:[1-9]\d*$/.test(batch.batchKey || '') || id(batch.customerId) !== id(candidate.customerId)
|| !Array.isArray(batch.sources) || !batch.sources.length || batch.sources.some(s => !/^(BOOKING:.+|ORDER:[1-9]\d*)$/.test(s.sourceKey || '') || !s.destinationKey?.startsWith('FBA:') || !amount(s.packageCount))
|| batch.sources.reduce((sum, s) => sum + Number(s.packageCount), 0) !== Number(candidate.availablePackages)) throw new DispatchNotRun('FBA批次未完整覆盖本托真实逐票货物')
let batchVersion
try { batchVersion = version(batch.recordVersion) } catch { throw new DispatchNotRun('FBA批次缺少权威recordVersion') }
for (const source of batch.sources) batchAllocations.push({ planPalletId: candidate.selection.planPalletId, batchKey: batch.batchKey,
customerId: candidate.customerId, sourceKey: source.sourceKey, destinationKey: source.destinationKey,
packageCount: Number(source.packageCount), recordVersion: batchVersion })
}
return { idempotencyKey: commandKey(marker, namespace, 'create'), warehouseId: candidates[0].warehouseId,
vehicle: null, driver: null, plannedDepartureAt: null, estimatedArrivalAt: null,
remark: `${marker}-${namespace}`, cargo, batchAllocations }
}
export function markerCandidates(context, candidates) {
const fixture = context.createdResources?.phase4OverseasPda
const pallets = new Set([...(fixture?.palletIds || []), ...Object.values(fixture?.generatedPalletsByInbound || {}).flat().filter(p => id(p.warehouseId) === id(fixture?.overseasWarehouseId)).map(p => p.palletId)].map(id))
if (fixture?.marker !== context.marker || fixture?.finalStatus !== 'COMPLETED' || !positiveId(fixture?.customerId) || !positiveId(fixture?.overseasWarehouseId)) throw new DispatchNotRun('缺少本轮PDA完成回执及真实客户仓')
return candidates.filter(candidate => {
try { sourceSelection(candidate) } catch { return false }
return candidate.selection.sourceType === 'INBOUND_PLAN' && pallets.has(id(candidate.selection.warehousePalletId))
&& id(candidate.customerId) === id(fixture.customerId) && id(candidate.warehouseId) === id(fixture.overseasWarehouseId)
&& candidate.destination.type === 'FBA'
})
}
export function notRun(context, proof, reason) {
proof.status = 'NOT_RUN'; proof.reason = reason
// 未运行时保留既有终态;否则显式标记 NOT_RUN,便于下游区分前置不足与代码失败。
proof.finalStatus ??= 'NOT_RUN'
context.addCoverageGap?.('phase4-overseas-dispatch', 'native-dispatch', '缺少当前生产前置', reason, 'fixture', { deploymentBlocker: true })
return proof
}
// 请求意图必须先持久化,未知写结果使同marker后续写入全部停止。
export function dispatchIo(context, module, proof, namespace) {
context.createdResources ??= {}; proof.requests ??= []
const boundary = context.createdResources.dispatchWriteBoundary ??= { marker: context.marker, stopped: false }
if (!context.marker || boundary.marker !== context.marker) throw new DispatchNotRun('请求边界不属于本轮marker')
const checkpoint = async () => { if (typeof context.checkpoint === 'function') await context.checkpoint() }
const check = (label, response, passed, detail = '') => {
context.assert?.(module, label, response, passed, `${detail ? `${detail}; ` : ''}marker=${context.marker}; HTTP=${response?.http}; requestId=${response?.requestId}; traceId=${response?.traceId}`, { scenario: namespace, deploymentBlocker: true })
if (!passed) throw Error(`${namespace}: ${label}`)
}
const send = async (method, path, body, label) => {
const write = !['GET', 'HEAD'].includes(method)
if (write && boundary.stopped) throw Error('本轮写请求结果未知或回执存档失败,停写并保留原键')
if (write && typeof context.checkpoint !== 'function') throw new DispatchNotRun('缺少真实持久化checkpoint,未发送排车写请求')
const entry = { method, path, ...(body === undefined ? {} : { body: structuredClone(body) }), label, status: 'SENDING' }
proof.requests.push(entry)
try { await checkpoint() } catch (error) { boundary.stopped = true; entry.status = 'NOT_SENT'; throw error }
let response
try { response = await context.request(method, path, body) }
catch (error) { entry.status = 'UNKNOWN_RESULT'; entry.error = error.message; if (write) { boundary.stopped = true; boundary.outcomeUnknown = true; proof.outcomeUnknown = true; proof.cleanup = 'PRESERVED_FOR_REVIEW' }; await checkpoint(); throw error }
const unknown = !response?.http || response.http >= 500 || write && (response.body?.code == null || success(response) && response.body.data == null)
Object.assign(entry, { status: unknown ? 'UNKNOWN_RESULT' : 'RESPONDED', http: response?.http,
code: response?.body?.code, message: response?.body?.message, raw: response?.raw, transportError: response?.transportError,
requestId: response?.requestId, traceId: response?.traceId, result: structuredClone(response?.body?.data) })
if (write && entry.status === 'UNKNOWN_RESULT') { boundary.stopped = true; boundary.outcomeUnknown = true; proof.outcomeUnknown = true; proof.cleanup = 'PRESERVED_FOR_REVIEW' }
try { await checkpoint() } catch (error) { if (write) boundary.stopped = true; throw error }
return response
}
const command = async (method, path, body, label) => { const response = await send(method, path, body, label); check(label, response, success(response)); return response.body.data }
const key = action => commandKey(context.marker, namespace, action)
return { check, send, command, key,
planIds: (context.createdResources?.phase4OverseasPda?.generatedPlanIds || []).map(String) }
}
export function replayFacts(result) { const copy = structuredClone(result); if (copy && typeof copy === 'object') delete copy.replayed; return copy }
export async function replay(io, path, body, result, label, method = 'POST') {
const again = await io.command(method, path, body, label)
io.check(`${label}恢复原命令结果`, null, canonical(replayFacts(again)) === canonical(replayFacts(result)))
return again
}
export async function candidatePage(io, warehouseId, extra = {}) {
const planIds = extra.planIds ?? io?.planIds
const params = { warehouseId, sourceStage: 'RECEIVED', ...(Array.isArray(planIds) && planIds.length ? { planIds } : {}), ...extra }
const result = await io.command('GET', query(`${DISPATCH_BASE}/candidates`, params), undefined, '读取同仓原生候选')
io.check('候选完整包络', null, result?.complete === true && Array.isArray(result.candidates))
return result.candidates
}
export async function readDispatch(io, dispatchNo) {
const path = `${DISPATCH_BASE}/${encodeURIComponent(dispatchNo)}`
const data = await io.command('GET', path, undefined, '回查原生整车详情')
const items = await io.command('GET', `${path}/items`, undefined, '回查真实货物明细')
const route = await io.command('GET', `${path}/route`, undefined, '回查真实起点与目的地')
const page = await io.command('GET', query(DISPATCH_BASE, { page: 1, size: 20, dispatchNo }), undefined, '按业务号核对整车仓库')
const rows = page?.records?.filter(r => r.dispatchNo === dispatchNo) || []
io.check('主单、货物、路线版本与实际仓一致', null, data?.dispatchNo === dispatchNo && items?.dispatchNo === dispatchNo && route?.dispatchNo === dispatchNo
&& rows.length === 1 && positiveId(rows[0].warehouseId) && id(version(data.recordVersion)) === id(version(items.recordVersion))
&& id(version(route.recordVersion)) === id(version(data.recordVersion)) && canonical(rows[0].counts) === canonical(data.counts)
&& id(rows[0].recordVersion) === id(data.recordVersion) && rows[0].status === data.status && rows[0].stage === data.stage
&& items.status === data.status && items.stage === data.stage)
return { data, items, route, warehouseId: rows[0].warehouseId }
}
export function dispatchMatches(bundle, candidates) {
const items = bundle?.items?.items, counts = bundle?.data?.counts
if (!Array.isArray(items) || !counts || !Array.isArray(candidates) || bundle.data.status !== 'ACTIVE' || bundle.data.stage !== 'BEFORE_LOADING'
|| items.length !== candidates.length || Number(counts.itemCount) !== items.length || Number(counts.palletCount) !== new Set(candidates.map(c => id(c.selection.warehousePalletId))).size
|| Number(counts.packageCount) !== candidates.reduce((n, c) => n + Number(c.availablePackages), 0) || Number(counts.pickedPackageCount) !== 0 || Number(counts.shippedPackageCount) !== 0) return false
const details = bundle.data.orderDetails
if (!Array.isArray(details) || new Set(details.map(d => d.sourceKey)).size !== candidates.length
|| details.some(d => !candidates.some(c => c.sourceKey === d.sourceKey && id(c.customerId) === id(d.customerId)
&& c.inboundCode === d.inboundCode && c.destination.warehouseCode === d.destinationFbaWarehouse))) return false
return new Set(items.map(i => id(i.dispatchItemId))).size === items.length && new Set(items.map(i => i.sourceKey)).size === items.length && items.every(i => positiveId(i.dispatchItemId) && candidates.some(c =>
c.sourceKey === i.sourceKey && c.sourceVersion === i.sourceVersion && id(c.customerId) === id(i.customerId) && id(c.warehouseId) === id(bundle.warehouseId)
&& id(c.selection.warehousePalletId) === id(i.warehousePalletId) && id(c.selection.planPalletId) === id(i.planPalletId)
&& c.inboundCode === i.inboundCode && c.destination.key === i.destination?.key && Number(c.availablePackages) === Number(i.scheduledPackages) && Number(i.pickedPackages) === 0))
}
export function groupedDispatchMatches(result, groups, candidates) {
if (!Array.isArray(result?.vehicles) || result.vehicles.length !== groups.length) return false
const seen = new Set(), numbers = new Set()
return result.vehicles.every(row => {
const group = groups.find(g => id(g.draftContainerId) === id(row.draftContainerId)), command = row.dispatch
const selected = candidates.filter(c => group?.sourceKeys.includes(c.sourceKey))
if (!group || seen.has(id(row.draftContainerId)) || numbers.has(command?.dispatchNo) || !command?.dispatchNo || command.action !== 'CREATE' || command.status !== 'ACTIVE' || command.stage !== 'BEFORE_LOADING') return false
seen.add(id(row.draftContainerId)); numbers.add(command.dispatchNo)
return Number(command.counts?.itemCount) === selected.length && Number(command.counts?.palletCount) === selected.length
&& Number(command.counts?.packageCount) === selected.reduce((n, c) => n + Number(c.availablePackages), 0)
})
}
export async function newDraft(io, warehouseId) {
const body = { idempotencyKey: io.key('draft-create'), warehouseId }
const draft = await io.command('POST', `${DISPATCH_BASE}/drafts`, body, '创建服务器草稿token')
io.check('服务器新草稿身份及零版本', null, Boolean(draft?.draftToken) && id(draft.warehouseId) === id(warehouseId) && id(version(draft.recordVersion)) === '0' && Array.isArray(draft.containers) && Array.isArray(draft.sources))
await replay(io, `${DISPATCH_BASE}/drafts`, body, draft, '创建草稿同键重放')
return draft
}
export async function activeDraft(io, token) {
const rows = await io.command('GET', `${DISPATCH_BASE}/drafts`, undefined, '回查当前拥有者草稿')
io.check('原服务器草稿唯一存在', null, Array.isArray(rows) && rows.filter(d => d.draftToken === token).length === 1)
return rows.find(d => d.draftToken === token)
}
// 分组全成功才移除已证明属于本轮的未拣实物;失败或未知响应保留所有已成功资源。
export async function verifyGroupedDispatch(context, module, candidates) {
const proof = { marker: context.marker, status: 'CHECKING', groups: [], results: [], cleanup: 'PRESERVED_FOR_REVIEW', requests: [] }
if (context.createdResources?.groupedDispatch?.requests?.length) throw Error('本轮分组尝试已经存在,禁止自动重造')
context.createdResources ??= {}; context.createdResources.groupedDispatch = proof
try {
const owned = markerCandidates(context, candidates)
if (owned.length !== candidates.length || candidates.length < 2 || new Set(candidates.map(c => c.sourceKey)).size !== candidates.length
|| new Set(candidates.map(c => id(c.selection?.warehousePalletId))).size !== candidates.length) throw new DispatchNotRun('分组须有本轮两个独立未拣实托')
candidates.forEach((c, n) => createBody(context.marker, `GROUP-${n}`, [c]))
const io = dispatchIo(context, module, proof, 'GROUPED')
let draft = await newDraft(io, candidates[0].warehouseId); proof.draftToken = draft.draftToken
for (const [n, candidate] of candidates.entries()) {
const previous = new Set(draft.containers.map(c => id(c.draftContainerId)))
draft = await io.command('POST', `${DISPATCH_BASE}/drafts/containers`, { recordVersion: version(draft.recordVersion), idempotencyKey: io.key(`container-${n}`), draftToken: draft.draftToken, capacity: 15 }, '创建真实草稿容器')
const additions = draft.containers.filter(c => !previous.has(id(c.draftContainerId)))
io.check('容器身份来自服务器草稿', null, additions.length === 1 && positiveId(additions[0].draftContainerId))
const group = { draftContainerId: additions[0].draftContainerId, sourceKeys: [candidate.sourceKey] }; proof.groups.push(group)
const body = { recordVersion: version(draft.recordVersion), idempotencyKey: io.key(`lock-${n}`), draftToken: draft.draftToken, draftContainerId: group.draftContainerId, source: sourceSelection(candidate) }
draft = await io.command('POST', `${DISPATCH_BASE}/draft-locks/lock`, body, '锁定完整真实来源到原容器')
await replay(io, `${DISPATCH_BASE}/draft-locks/lock`, body, draft, '锁定同键重放')
}
const snapshot = { recordVersion: version(draft.recordVersion), idempotencyKey: io.key('snapshot'), draftToken: draft.draftToken, planIds: [], containers: draft.containers }
draft = await io.command('POST', `${DISPATCH_BASE}/drafts/snapshot`, snapshot, '保存版本化完整容器布局')
await replay(io, `${DISPATCH_BASE}/drafts/snapshot`, snapshot, draft, '布局同键重放')
const groups = proof.groups.map((g, n) => ({ draftContainerId: g.draftContainerId, dispatch: createBody(context.marker, `GROUPED-CHILD-${n}`, candidates.filter(c => g.sourceKeys.includes(c.sourceKey))) }))
const before = await activeDraft(io, draft.draftToken)
const wrong = { recordVersion: version(draft.recordVersion), idempotencyKey: io.key('convert-reject'), draftToken: draft.draftToken,
groups: groups.map((g, n) => ({ ...g, dispatch: groups[(n + 1) % groups.length].dispatch })) }
const rejected = await io.send('POST', `${DISPATCH_BASE}/drafts/pre-arrange`, wrong, '错组转换拒绝')
io.check('错组来源明确冲突', rejected, [200, 409].includes(rejected?.http) && String(rejected.body?.code) === 'A0409')
io.check('拒绝后原草稿不变', null, canonical(await activeDraft(io, draft.draftToken)) === canonical(before))
const body = { recordVersion: version(draft.recordVersion), idempotencyKey: io.key('convert'), draftToken: draft.draftToken, groups }
const converted = await io.command('POST', `${DISPATCH_BASE}/drafts/pre-arrange`, body, '原子转换一容器一整车')
io.check('真实分组命令数量与货物核平', null, converted?.draftToken === draft.draftToken && groupedDispatchMatches(converted, proof.groups, candidates))
proof.results = converted.vehicles
for (const row of proof.results) {
const selected = candidates.filter(c => proof.groups.find(g => id(g.draftContainerId) === id(row.draftContainerId)).sourceKeys.includes(c.sourceKey))
const current = await readDispatch(io, row.dispatch.dispatchNo)
io.check('成功创建后逐车回查真实来源', null, dispatchMatches(current, selected))
}
await replay(io, `${DISPATCH_BASE}/drafts/pre-arrange`, body, converted, '整批转换同键重放')
for (const row of proof.results) {
let current = await readDispatch(io, row.dispatch.dispatchNo)
const selected = candidates.filter(c => proof.groups.find(g => id(g.draftContainerId) === id(row.draftContainerId)).sourceKeys.includes(c.sourceKey))
io.check('清理前仍为本轮完整未拣货物', null, dispatchMatches(current, selected))
for (const item of current.items.items) {
io.check('服务端允许移出本轮实物', null, item.remove?.allowed === true)
const remove = { recordVersion: version(current.data.recordVersion), idempotencyKey: io.key(`remove-${item.dispatchItemId}`), reason: `${context.marker}-分组验收释放未拣实物` }
await io.command('POST', `${DISPATCH_BASE}/${encodeURIComponent(row.dispatch.dispatchNo)}/items/${encodeURIComponent(item.dispatchItemId)}/remove`, remove, '版本化移出本轮分组实物')
current = await readDispatch(io, row.dispatch.dispatchNo)
}
io.check('空对照单真实数量归零', null, dispatchMatches(current, []))
}
const available = await candidatePage(io, candidates[0].warehouseId)
io.check('原来源回到可选集合', null, candidates.every(c => available.some(a => a.sourceKey === c.sourceKey && a.blockReasons?.length === 0)))
proof.cleanup = 'OWNED_UNPICKED_SOURCES_RELEASED'; proof.status = 'GROUPED_DISPATCH_VERIFIED'; await context.checkpoint()
return proof
} catch (error) {
if (error.notRun && !proof.requests.length) return notRun(context, proof, error.message)
proof.status = proof.outcomeUnknown ? 'OUTCOME_UNKNOWN' : 'UNPROVEN'; proof.error = error.message
throw error
}
}