77 lines
7.2 KiB
JavaScript
77 lines
7.2 KiB
JavaScript
import { createHash } from 'node:crypto'
|
|
import { isDeepStrictEqual } from 'node:util'
|
|
|
|
export const deliveryFaultScopeKeys = ['marker','customerId','deliveryId','sourceInboundId','sourceOrderId','rootOrderId']
|
|
const canonical = value => Array.isArray(value) ? value.map(canonical) : value && typeof value === 'object'
|
|
? Object.fromEntries(Object.keys(value).sort().map(key => [key, canonical(value[key])])) : value
|
|
export function assertTestDeliveryFaultProof(proof, scope) {
|
|
const need = (value, code) => { if (!value) throw Error('Invalid delivery fault proof: ' + code) }
|
|
need(proof?.readOnly === true && deliveryFaultScopeKeys.every(key => proof[key] === scope[key]), 'SCOPE')
|
|
const row=proof.producer, before=proof.beforeRetry, after=proof.afterRetry, env=row?.envelope, payload=env?.payload
|
|
need(row && /^[1-9][0-9]{0,19}$/.test(row.entryId) && /^(?:[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26})$/.test(row.eventId), 'IDENTITY')
|
|
need(row.eventType==='OverseasDeliveryCompleted' && row.customerId===scope.customerId && row.aggregateType==='INBOUND_ORDER'
|
|
&& row.aggregateId===scope.sourceInboundId && payload?.deliveryOrderId===scope.deliveryId
|
|
&& payload.sourceInboundId===scope.sourceInboundId && payload.sourceOrderId===scope.sourceOrderId
|
|
&& payload.rootOrderId===scope.rootOrderId && payload.deliveryNo && payload.inboundCode && payload.sourceOrderNo
|
|
&& payload.fbaWarehouseCode && payload.lastlegOrderId && payload.lastlegOrderNo && Number.isFinite(Date.parse(payload.completedAt))
|
|
&& row.eventKey===`OverseasDeliveryCompleted:${payload.deliveryNo}:${scope.sourceInboundId}`, 'PRODUCER')
|
|
need(env.eventId===row.eventId && env.eventType===row.eventType && env.tenantId===scope.customerId
|
|
&& Number(env.eventVersion)===1 && env.aggregateType==='INBOUND_ORDER' && env.aggregateId===scope.sourceInboundId
|
|
&& Number(env.aggregateVersion)>0, 'ENVELOPE')
|
|
const hash='sha256:'+createHash('sha256').update(JSON.stringify(canonical(payload))).digest('hex')
|
|
need(row.payloadHash===hash && env.payloadHash===hash, 'HASH')
|
|
need(before?.publishStatus==='DEAD' && before.failedAttempts===12 && row.publishStatus==='PUBLISHED'
|
|
&& after?.publishStatus==='PUBLISHED' && [before,after].every(value => value.entryId===row.entryId
|
|
&& value.eventId===row.eventId && value.payloadHash===hash && isDeepStrictEqual(value.envelope,env)), 'ORIGINAL_RETRY')
|
|
const inbox=proof.consumer
|
|
need(inbox?.count===1 && inbox.eventId===row.eventId && inbox.eventType===row.eventType && inbox.aggregateId===scope.sourceInboundId
|
|
&& inbox.status==='PROCESSED' && inbox.payloadHash===hash && isDeepStrictEqual(inbox.payload,payload), 'INBOX')
|
|
const attempts=proof.attempts, matches=proof.proxyRecords?.filter(value=>value.matched===true)
|
|
need(Array.isArray(attempts) && attempts.length===13 && new Set(attempts.map(value=>value.attemptId)).size===13
|
|
&& attempts.every((value,index)=>value.entryId===row.entryId && value.eventId===row.eventId && value.serviceName==='lastleg'
|
|
&& value.outcome===(index<11?'RETRY':index===11?'DEAD':'DUPLICATE')), 'ATTEMPTS')
|
|
need(matches?.length===13 && matches.every((value,index)=>value.eventId===row.eventId && value.payloadHash===hash
|
|
&& value.attempt===index+1 && value.outcome===(index===0?'APPLIED':'DUPLICATE')
|
|
&& (index<12?value.errorCode==='ACK_LOSS_INJECTED':value.errorCode===null)), 'REAL_ACK_LOSS')
|
|
const audit=proof.retryAudit
|
|
need(proof.retryAuditCount===1 && audit?.requestId===proof.retryRequestId && audit.entryId===row.entryId
|
|
&& audit.eventId===row.eventId && audit.payloadHash===hash && audit.previousFailures===12, 'RETRY_AUDIT')
|
|
const facts=before.business
|
|
need(facts && isDeepStrictEqual(facts,after.business), 'BUSINESS_UNCHANGED')
|
|
for (const key of ['inbound','inboundHistory','inbox','order','orderHistory','completionEvent']) need(Array.isArray(facts[key]), 'BUSINESS_SHAPE')
|
|
need(facts.inbox.length===1 && facts.inbox[0].eventId===row.eventId && facts.inbox[0].status==='PROCESSED'
|
|
&& facts.inbox[0].payloadHash===hash && facts.inbox[0].processedAt && isDeepStrictEqual(facts.inbox[0].payload,payload), 'PERSISTED_INBOX')
|
|
const inbounds=facts.inbound.filter(value=>value.id===scope.sourceInboundId)
|
|
need(inbounds.length===1 && inbounds[0].customerId===scope.customerId && inbounds[0].orderId===scope.sourceOrderId
|
|
&& inbounds[0].orderNo===payload.sourceOrderNo && inbounds[0].inboundCode===payload.inboundCode
|
|
&& inbounds[0].processStatus==='DELIVERED' && Number.isInteger(inbounds[0].version), 'INBOUND')
|
|
need(facts.inboundHistory.filter(value=>value.inboundId===scope.sourceInboundId && value.action==='COMPLETE_DELIVERY'
|
|
&& value.fromStatus==='LOADED' && value.toStatus==='DELIVERED').length===1, 'INBOUND_HISTORY')
|
|
const instant=payload.completedAt.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z$/)
|
|
const completedMillis=instant?Date.parse(instant[1]+'Z')+Math.floor((Number((instant[2]||'').padEnd(9,'0'))+500000)/1000000):NaN
|
|
const storedMillis=value=>Date.parse(String(value).replace(' ','T').replace(/Z$/,'')+'Z')
|
|
need(facts.inboundHistory.filter(value=>value.action==='COMPLETE_DELIVERY')
|
|
.every(value=>storedMillis(value.occurredAt)===completedMillis), 'COMPLETION_TIME')
|
|
need(facts.order.length>0 && new Set(facts.order.map(value=>value.id)).size===facts.order.length
|
|
&& facts.order.every(value=>value.customerId===scope.customerId && value.rootOrderId===scope.rootOrderId && Number.isInteger(value.version)), 'ORDER_SCOPE')
|
|
const source=facts.order.find(value=>value.id===scope.sourceOrderId), root=facts.order.find(value=>value.id===scope.rootOrderId)
|
|
need(source && root && root.orderRole==='MASTER' && source.orderNo===payload.sourceOrderNo && source.inboundCode===payload.inboundCode, 'ORDER_IDENTITY')
|
|
const historyCount=(id,action)=>facts.orderHistory.filter(value=>value.orderId===id && value.action===action).length
|
|
need(historyCount(source.id,'OVERSEAS_DELIVERED')===1, 'SOURCE_SIGNED')
|
|
need(facts.orderHistory.filter(value=>value.orderId===source.id && value.action==='OVERSEAS_DELIVERED')
|
|
.every(value=>storedMillis(value.occurredAt)===completedMillis), 'SOURCE_SIGNED_TIME')
|
|
const rootCanComplete=historyCount(root.id,'OVERSEAS_DELIVERED')===1
|
|
&& facts.order.filter(value=>value.id!==root.id && value.status!=='CANCELLED').every(value=>value.status==='COMPLETED')
|
|
need(rootCanComplete ? root.status==='COMPLETED' : ['APPROVED','ACCEPTED','PROCESSING'].includes(root.status), 'ROOT_GATE')
|
|
need(source.id===root.id || source.status==='COMPLETED', 'SOURCE_COMPLETION')
|
|
need(source.status==='COMPLETED'?inbounds[0].orderStatusSnapshot==='COMPLETED':inbounds[0].orderStatusSnapshot!=='COMPLETED', 'INBOUND_ORDER_SNAPSHOT')
|
|
for (const order of facts.order) {
|
|
const events=facts.completionEvent.filter(value=>value.orderId===order.id)
|
|
need(events.length===(order.status==='COMPLETED'?1:0) && events.every(value=>value.status==='PUBLISHED'
|
|
&& value.eventType==='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED'), 'COMPLETION_EVENTS')
|
|
if(order.status==='COMPLETED') need(order.completedAt && historyCount(order.id,order.id===root.id?'FULFILLMENT_COMPLETED':'OVERSEAS_DELIVERED')===1, 'COMPLETION_HISTORY')
|
|
}
|
|
need(facts.completionEvent.every(value=>facts.order.some(order=>order.id===value.orderId)), 'FOREIGN_EVENT')
|
|
return proof
|
|
}
|