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 } }