baiyue-firstleg/scripts/test-event-fault-executor.mjs

// Internal single TEST fault lifecycle. Uses the existing ACK-loss proxy and
// authenticated original-DEAD retry endpoint, never direct business SQL writes.
import { readFileSync } from 'node:fs'
import { createHash } from 'node:crypto'
import { isDeepStrictEqual } from 'node:util'
import { assertTestDeliveryFaultProof, deliveryFaultScopeKeys } from './test-delivery-fault-proof.mjs'
import { assertTestFeeVoidFaultProof, feeVoidScopeKeys, feeSourceTypes } from './test-fee-void-fault-proof.mjs'
export { assertTestFeeVoidFaultProof } from './test-fee-void-fault-proof.mjs'
export { assertTestDeliveryFaultProof } from './test-delivery-fault-proof.mjs'
import { validateEvidenceConfig } from './fba-core-regression-evidence.mjs'

const runtime = readFileSync(new URL('./test-event-fault-runtime.py', import.meta.url), 'utf8')
const proxySource = readFileSync(new URL('./event-delivery-fault-proxy.py', import.meta.url)).toString('base64')
const encode = value => Buffer.from(value).toString('base64')
const expression = value => '[' + encode(value).match(/.{1,4000}/g).map(part => `'${part}'`).join(',') + "].join('')"

export function renderTestFaultAction(action, scope, agent) {
  return renderAction(action,scope,agent,'LOADING')
}
export function renderTestUnloadFaultAction(action, scope, agent) {
  return renderAction(action,scope,agent,'UNLOAD')
}
function renderAction(action,scope,agent,direction) {
  const keys=direction==='FEE_VOID'?feeVoidScopeKeys:direction==='DELIVERY'?deliveryFaultScopeKeys:direction==='UNLOAD'?['marker','customerId','containerBookingId']:['marker','customerId','loadingPlanId','containerBookingId']
  if (!(action==='preflight'&&direction==='FEE_VOID')&&!['prepare', 'activate', 'proxy-ready', 'health', 'status', 'recover', 'proof', 'restore', 'finish'].includes(action)
      || !/^[A-Za-z0-9_.-]{1,128}$/.test(agent) || !/^[A-Za-z0-9_-]{1,96}$/.test(scope?.marker || '')
      || keys.filter(key=>!['marker','sourceType'].includes(key)).some(key => !/^[1-9][0-9]{0,19}$/.test(scope?.[key] || ''))
      || (direction==='FEE_VOID'&&(!feeSourceTypes.includes(scope?.sourceType)||Number(scope?.sourceVersion)>2147483647))
      || (direction==='FEE_VOID'&&keys.some(key=>typeof scope?.[key]!=='string'))
      || (direction!=='LOADING'&&Object.keys(scope).some(key=>!keys.includes(key)))) {
    throw Error('Invalid TEST fault action or scope')
  }
  const config = encode(JSON.stringify({ action, scope, direction, ...(action === 'prepare' ? { proxySource } : {}) }))
  const groovy = `def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${expression(runtime)}),'UTF-8'),'${config}']).redirectErrorStream(true).start();def output=p.inputStream.getText('UTF-8');p.waitFor();print(output)`
  return `def c=jenkins.model.Jenkins.instance.getComputer('${agent}');if(c==null || c.channel==null)throw new RuntimeException('TEST fault agent unavailable');print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${expression(groovy)}),'UTF-8'),c.channel))`
}

export function createTestFaultExecutor(env, fetcher = globalThis.fetch) {
  return createExecutor(env,fetcher,'LOADING')
}
export function createTestDeliveryFaultExecutor(env, fetcher = globalThis.fetch) {
  return createExecutor(env,fetcher,'DELIVERY')
}
export function createTestFeeVoidFaultExecutor(env, fetcher = globalThis.fetch) {
  return createExecutor(env,fetcher,'FEE_VOID')
}
export function createTestUnloadFaultExecutor(env, fetcher = globalThis.fetch) {
  return createExecutor(env,fetcher,'UNLOAD')
}
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 mysqlDatetimeMillisFromInstant(value) {
  const match=typeof value==='string'&&value.match(/^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2})(?:\.(\d{1,9}))?Z$/)
  if(!match)return NaN
  const seconds=Date.parse(match[1]+'Z')
  const nanos=Number((match[2]||'').padEnd(9,'0'))
  // DATETIME(3) rounds fractional seconds; Date.parse alone truncates them.
  return seconds+Math.floor((nanos+500000)/1000000)
}
export function assertTestUnloadFaultProof(proof,scope) {
  const require=(passed,code)=>{if(!passed)throw Error('Invalid unload fault proof: '+code)}
  require(proof?.readOnly===true&&Object.keys(scope).every(key=>proof[key]===scope[key]),'SCOPE')
  const before=proof.beforeRetry,after=proof.afterRetry,row=proof.producer,env=row?.envelope
  require(row&&env&&/^[1-9][0-9]{0,19}$/.test(row.entryId)&&/^(?:[0-9a-fA-F-]{36}|[0-9A-HJKMNP-TV-Z]{26})$/.test(row.eventId),'IDENTITY')
  require(row.eventType==='OverseasContainerUnloaded'&&row.customerId==='0'&&row.aggregateId===scope.containerBookingId
    &&row.eventKey===`OverseasContainerUnloaded:${scope.containerBookingId}`&&row.aggregateType==='CONTAINER_BOOKING','PRODUCER')
  require(env.eventId===row.eventId&&env.eventType===row.eventType&&Number(env.eventVersion)===1&&env.tenantId==='0'
    &&env.aggregateId===scope.containerBookingId&&env.aggregateType==='CONTAINER_BOOKING'
    &&env.payload?.sourceContainerBookingId===scope.containerBookingId&&Boolean(env.payload.containerNo),'ENVELOPE')
  const hash='sha256:'+createHash('sha256').update(JSON.stringify(canonical(env.payload))).digest('hex')
  require(row.payloadHash===hash&&env.payloadHash===hash,'HASH')
  require(before?.publishStatus==='DEAD'&&before.failedAttempts===12&&row.publishStatus==='PUBLISHED'
    &&before.entryId===row.entryId&&before.eventId===row.eventId&&before.payloadHash===hash
    &&isDeepStrictEqual(before.envelope,env)&&after?.publishStatus==='PUBLISHED'&&after.entryId===row.entryId
    &&after.eventId===row.eventId&&after.payloadHash===hash&&isDeepStrictEqual(after.envelope,env),'ORIGINAL_RETRY')
  const inbox=proof.consumer
  require(inbox?.count===1&&inbox.status==='PROCESSED'&&inbox.eventId===row.eventId&&inbox.payloadHash===hash
    &&inbox.eventType===row.eventType&&inbox.aggregateId===scope.containerBookingId&&isDeepStrictEqual(inbox.payload,env.payload),'INBOX')
  const attempts=proof.attempts||[],proxy=proof.proxyRecords||[]
  require(attempts.length===13&&new Set(attempts.map(a=>a.attemptId)).size===13&&attempts.every(a=>a.eventId===row.eventId&&a.entryId===row.entryId&&a.serviceName==='lastleg')
    &&attempts.slice(0,11).every(a=>a.outcome==='RETRY')&&attempts[11].outcome==='DEAD'&&attempts[12].outcome==='DUPLICATE','ATTEMPTS')
  const matches=proxy.filter(p=>p.matched===true)
  require(matches.length===13&&matches.every((p,i)=>p.eventId===row.eventId&&p.payloadHash===hash&&p.attempt===i+1)
    &&matches.slice(0,12).every((p,i)=>p.errorCode==='ACK_LOSS_INJECTED'&&p.outcome===(i===0?'APPLIED':'DUPLICATE'))
    &&matches[12].outcome==='DUPLICATE'&&matches[12].errorCode==null,'REAL_ACK_LOSS')
  const audit=proof.retryAudit
  require(audit&&proof.retryAuditCount===1&&audit.requestId===proof.retryRequestId&&audit.entryId===row.entryId
    &&audit.eventId===row.eventId&&audit.payloadHash===hash&&audit.previousFailures===12,'RETRY_AUDIT')
  const container=before.container
  require(container&&container.id===scope.containerBookingId&&container.containerNo===env.payload.containerNo
    &&isDeepStrictEqual(container,after.container)&&container.version!=null&&container.emptyReportSourceVersion!=null,'CONTAINER_UNCHANGED')
  const time=value=>Date.parse(String(value).replace(' ','T').replace(/Z$/,'')+'Z')
  require(time(container.unloadingCompletedAt)===mysqlDatetimeMillisFromInstant(env.payload.unloadingCompletedAt)
    &&time(container.unloadingPhotoUploadedAt)===mysqlDatetimeMillisFromInstant(env.payload.unloadingPhotoUploadedAt)
    &&time(container.estimatedEmptyReportedAt)-time(container.unloadingCompletedAt)===72*3600000
    &&(env.payload.actualEmptyReportedAt==null?container.actualEmptyReportedAt===null:time(container.actualEmptyReportedAt)===Date.parse(env.payload.actualEmptyReportedAt))
    &&String(container.emptyReportSourceVersion)===String(env.aggregateVersion),'CONTAINER_FACTS')
  return proof
}
function createExecutor(env,fetcher,direction) {
  validateEvidenceConfig(env)
  if (env.CHONGCHONG_ENV !== 'test') throw Error('TEST fault requires CHONGCHONG_ENV=test')
  let scope = null, finished = false, preparedEvidence=null, restoration=null, preparationAttempted=false, ready=false
  async function action(name) {
    const script = renderAction(name, scope, env.FBA_CORE_JENKINS_AGENT,direction)
    const response = await fetcher(new URL('scriptText', env.JENKINS_URL.replace(/\/?$/, '/')), {
      method: 'POST', redirect: 'error', headers: { Authorization: 'Basic ' + encode(env.JENKINS_USER + ':' + env.JENKINS_API_TOKEN),
        'Content-Type': 'application/x-www-form-urlencoded' }, body: new URLSearchParams({ script }), signal: AbortSignal.timeout(55000) })
    if (!response.ok) throw Error('TEST fault HTTP ' + response.status)
    const rows = (await response.text()).split(/\r?\n/).filter(row => row.startsWith('TEST_EVENT_JSON='))
    if (rows.length !== 1) throw Error('TEST fault evidence missing or ambiguous')
    const result = JSON.parse(rows[0].slice('TEST_EVENT_JSON='.length))
    if (!result?.ok) throw Error('TEST fault action failed: ' + (/^[A-Z_0-9]{1,80}$/.test(result?.error) ? result.error : 'UNKNOWN'))
    return result.data
  }
  async function wait(name, check, attempts = 120, tolerateStartup = false) {
    for (let index = 0; index < attempts; index++) {
      let value
      try { value = await action(name) } catch (error) {
        if (!tolerateStartup || !String(error.message).endsWith('TEST_EVENT_ACTION_FAILED')) throw error
      }
      if (value && check(value)) return value
      await new Promise(resolve => setTimeout(resolve, 1000))
    }
    throw Error('TEST fault observation timed out: ' + name)
  }
  async function cleanup() {
    if(direction==='FEE_VOID'&&!preparationAttempted)return {attempted:false}
    if (!scope || finished) return restoration
    await action('restore')
    await wait('health', row => row.healthy)
    restoration=await action('finish')
    if(direction!=='LOADING'&&!(/^sha256:[0-9a-f]{64}$/.test(preparedEvidence?.originalImageId||'')
      &&/^[0-9a-f]{64}$/.test(preparedEvidence?.originalFingerprint||'')
      &&restoration?.restored===true&&restoration.healthy===true&&restoration.proxyStopped===true
      &&restoration.producerName==='firstleg-test-lastleg'&&restoration.originalImageId===preparedEvidence?.originalImageId
      &&restoration.currentImageId===preparedEvidence?.originalImageId&&restoration.originalFingerprint===preparedEvidence?.originalFingerprint
      &&restoration.currentFingerprint===preparedEvidence?.originalFingerprint))throw Error((direction==='FEE_VOID'?'Fee void':direction==='DELIVERY'?'Delivery':'Unload')+' restoration evidence mismatch')
    finished = true
    return restoration
  }
  return {
    async prepare(next) {
      if (scope) {
        if(direction!=='LOADING'&&!isDeepStrictEqual(scope,next))throw Error('TEST unload fault scope drift')
        if(direction==='FEE_VOID'&&!ready)throw Error('Fee void preparation did not complete')
        return // Pin only the first actual current-run plan; other events pass through.
      }
      if(direction!=='LOADING')renderAction('prepare',next,env.FBA_CORE_JENKINS_AGENT,direction)
      scope = direction!=='LOADING'?{...next}:{ marker: next.marker, customerId: next.customerId,
        loadingPlanId: next.loadingPlanId, containerBookingId: next.containerBookingId }
      if(direction==='FEE_VOID')await wait('preflight', row=>row.ready===true,60)
      preparationAttempted=true
      preparedEvidence=await action('prepare')
      await wait('proxy-ready', row => row.ready, 10, true)
      await action('activate')
      await wait('health', row => row.healthy)
      ready=true
    },
    async readProof(expected) {
      if (!scope || (direction!=='LOADING'?!isDeepStrictEqual(scope,expected):Object.keys(scope).some(key => expected?.[key] !== scope[key]))) throw Error('TEST fault scope drift')
      try {
        await wait('status', row => row.producer?.publishStatus === 'DEAD' && row.producer.failedAttempts === 12, 180)
        await action('recover')
        await wait('status', row => row.producer?.publishStatus === 'PUBLISHED', 90)
        const proof = await action('proof')
        if(direction==='UNLOAD')assertTestUnloadFaultProof(proof,scope)
        if(direction==='DELIVERY')assertTestDeliveryFaultProof(proof,scope)
        if(direction==='FEE_VOID') {
          if(!isDeepStrictEqual(proof.prerequisite,preparedEvidence?.prerequisite))throw Error('Fee prerequisite evidence drift')
          assertTestFeeVoidFaultProof(proof,scope)
        }
        await cleanup()
        return { ...proof, restored: true, ...(direction!=='LOADING'?{restoration}:{}) }
      } finally { if(direction!=='LOADING')await cleanup() }
    },
    cleanup,
  }
}