baiyue-firstleg/scripts/fba-core-regression-evidence.mjs

// Internal explicit TEST evidence queries; no network or environment reads on import.
import { buildFleetInventorySql, fleetInventoryFromRows } from './fleet-inventory-evidence.mjs'
const stages = new Set(['BEFORE_COMPLETE', 'AFTER_COMPLETE', 'AFTER_SAME_KEY_REPLAY', 'AFTER_FRESH_KEY_REJECTION', 'AFTER_EVENT_RECOVERY'])
const numeric = new Set(['availablePackages', 'holdPackages', 'allocatedPackages', 'pickedPackages', 'inTransitPackages', 'receivingPackages', 'procurementPackages', 'receivedPackages', 'beforePackages', 'deltaPackages', 'afterPackages', 'packageCount'])
const fields = {
  sources: 'id customerId overseasWarehouseId sourceContainerBookingId containerNo sourceInboundId inboundCode sourceRegion packageCount',
  sessions: 'id sourceContainerBookingId containerNo overseasWarehouseId',
  balances: 'id customerId overseasWarehouseId destinationWarehouseId destinationWarehouseCode inboundOrderId inboundOrderNo inboundCode skuCode availablePackages holdPackages allocatedPackages pickedPackages inTransitPackages receivingPackages procurementPackages',
  positions: 'id balanceId customerId overseasWarehouseId destinationWarehouseCode palletId palletNo batchNo warehouseLocationId warehouseLocationCode status availablePackages holdPackages allocatedPackages pickedPackages receivingPackages',
  stockReceipts: 'id customerId planId taskId allocationId lineId orderId inboundCode skuCode warehousePalletId palletNo warehouseLocationId warehouseLocationCode batchNo receivedPackages inventoryBalanceId inventoryPositionId inventoryOperationKey',
  inventoryOperations: 'id customerId idempotencyKey operationType requestHash status',
  stockInLedger: 'id customerId balanceId positionId overseasWarehouseId inboundOrderId inboundCode palletId palletNo batchNo skuCode businessType businessOrderNo sourceEventId idempotencyKey statusFrom statusTo beforePackages deltaPackages afterPackages',
  receivingOperations: 'id sessionId action idempotencyKey requestHash',
}
const snake = value => value.replace(/[A-Z]/g, letter => '_' + letter.toLowerCase())
function identifier(value) {
  if (typeof value !== 'string' || !/^[1-9][0-9]{0,19}$/.test(value)) throw Error('Invalid receiving evidence identifier')
  return value
}
// UTF-8 hex literals avoid quote, backslash and SQL-mode ambiguity.
const literal = value => `(CONVERT(0x${Buffer.from(value, 'utf8').toString('hex')} USING utf8mb4) COLLATE utf8mb4_bin)`
function validateScope(scope) {
  for (const name of ['customerId', 'warehouseId', 'bookingId']) identifier(scope?.[name])
  if (!stages.has(scope.stage) || typeof scope.containerNo !== 'string' || !scope.containerNo.trim()
      || !Array.isArray(scope.inboundCodes) || !scope.inboundCodes.length
      || scope.inboundCodes.some(code => typeof code !== 'string' || !code.trim() || code.length > 128)
      || new Set(scope.inboundCodes).size !== scope.inboundCodes.length
      || !Array.isArray(scope.sourceInboundIds) || scope.sourceInboundIds.length !== scope.inboundCodes.length
      || new Set(scope.sourceInboundIds).size !== scope.sourceInboundIds.length) throw Error('Invalid receiving evidence scope')
  scope.sourceInboundIds.forEach(identifier)
  for (const code of scope.inboundCodes) if (!Number.isSafeInteger(scope.expectedPackagesByInbound?.[code])
      || scope.expectedPackagesByInbound[code] <= 0) throw Error('Invalid receiving evidence quantity')
}
export function buildReceivingEvidenceSql(scope) {
  validateScope(scope)
  const owned = alias => `${alias}.customer_id=${literal(scope.customerId)} AND ${alias}.overseas_warehouse_id=${literal(scope.warehouseId)}`
  const source = `o.customer_id=${literal(scope.customerId)} AND o.source_container_booking_id=${literal(scope.bookingId)}`
  const orderIds = `SELECT o.id FROM ll_order o WHERE ${source}`
  const session = `s.source_container_booking_id=${literal(scope.bookingId)} AND s.container_no=${literal(scope.containerNo)} AND s.overseas_warehouse_id=${scope.warehouseId}`
  const taskIds = `SELECT t.id FROM ll_inbound_pda_task t JOIN ll_inbound_plan p ON p.id=t.plan_id JOIN ll_order o ON o.id=p.source_order_id WHERE ${source} AND BINARY t.customer_id=BINARY o.customer_id AND BINARY p.customer_id=BINARY o.customer_id`
  const query = (name, alias, from, where) => {
    const pairs = fields[name].split(' ').map(field => {
      const column = `${alias}.${snake(field)}`
      return `'${field}',${numeric.has(field) ? column : `CAST(${column} AS CHAR)`}`
    }).join(',')
    return `SELECT JSON_OBJECT('collection','${name}','row',JSON_OBJECT(${pairs})) FROM ${from} WHERE ${where} ORDER BY ${alias}.id;`
  }
  return [
    'SET NAMES utf8mb4 COLLATE utf8mb4_bin;',
    'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
    'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;',
    query('sources', 'o', 'll_order o', source),
    query('sessions', 's', 'll_receiving_container_session s', session),
    query('balances', 'b', 'll_inventory_balance b', owned('b')),
    query('positions', 'p', 'll_inventory_position p', owned('p')),
    query('stockReceipts', 'r', 'll_inbound_stock_receipt r', `r.customer_id=${literal(scope.customerId)} AND r.order_id IN (${orderIds})`),
    query('inventoryOperations', 'i', 'll_inventory_operation i', `i.customer_id=${literal(scope.customerId)} AND i.operation_type=${literal('STOCK_IN')} AND EXISTS (SELECT 1 FROM ll_inbound_pda_task t WHERE t.id IN (${taskIds}) AND i.idempotency_key LIKE (CONCAT('PDA_STOCK_IN:',t.id,':%') COLLATE utf8mb4_bin))`),
    query('stockInLedger', 'l', 'll_inventory_ledger l', `${owned('l')} AND l.business_type='STOCK_IN' AND l.inbound_order_id IN (${orderIds})`),
    query('receivingOperations', 'r', 'll_receiving_container_operation r', `r.action='COMPLETE' AND r.session_id IN (SELECT s.id FROM ll_receiving_container_session s WHERE ${session})`),
    'COMMIT;',
  ].join('\n')
}
export function receivingEvidenceFromRows(scope, rows) {
  validateScope(scope)
  const result = Object.fromEntries(Object.keys(fields).map(name => [name, []]))
  if (!Array.isArray(rows)) throw Error('Receiving evidence rows missing')
  for (const item of rows) {
    if (!Object.hasOwn(result, item?.collection) || !item.row) throw Error('Invalid receiving evidence collection')
    result[item.collection].push(item.row)
  }
  const sources = result.sources
  if (sources.length !== scope.inboundCodes.length
      || new Set(sources.map(row => row.inboundCode)).size !== sources.length
      || new Set(sources.map(row => row.sourceInboundId)).size !== sources.length
      || sources.some(row => row.customerId !== scope.customerId || row.overseasWarehouseId !== scope.warehouseId
        || row.sourceContainerBookingId !== scope.bookingId || row.containerNo !== scope.containerNo
        || row.sourceRegion !== 'FIRSTLEG' || !scope.sourceInboundIds.includes(row.sourceInboundId)
        || !scope.inboundCodes.includes(row.inboundCode) || row.packageCount !== scope.expectedPackagesByInbound[row.inboundCode])
      || result.sessions.length !== 1 || result.sessions.some(row => row.sourceContainerBookingId !== scope.bookingId
        || row.containerNo !== scope.containerNo || row.overseasWarehouseId !== scope.warehouseId)) throw Error('Receiving evidence source/session scope mismatch')
  delete result.sources
  delete result.sessions
  for (const rows of Object.values(result)) rows.sort((a, b) => String(a.id).localeCompare(String(b.id)))
  result.palletComposition = result.positions.map(row => ({ ...row }))
  return result
}
export function createReceivingEvidenceReader(executeReadOnly) {
  if (typeof executeReadOnly !== 'function') throw Error('Receiving evidence transport missing')
  return async scope => receivingEvidenceFromRows(scope, await executeReadOnly(buildReceivingEvidenceSql(scope)))
}

const transactionStart = ['SET NAMES utf8mb4 COLLATE utf8mb4_bin;',
  'SET SESSION TRANSACTION ISOLATION LEVEL REPEATABLE READ;',
  'START TRANSACTION WITH CONSISTENT SNAPSHOT, READ ONLY;']
export function assertReadOnlySql(sql) {
  if (typeof sql !== 'string' || /--|\/\*|\b(?:INSERT|UPDATE|DELETE|REPLACE|DROP|ALTER|TRUNCATE|CALL|EXECUTE|PREPARE|INTO|OUTFILE|DUMPFILE|LOAD_FILE|SLEEP|BENCHMARK|GET_LOCK)\b/i.test(sql))
    throw Error('Read-only SQL rejected')
  const lines = sql.split('\n').filter(Boolean)
  if (!transactionStart.every((line,i) => lines[i] === line) || lines.at(-1) !== 'COMMIT;'
    || lines.length < 5 || lines.slice(3,-1).some(line => !/^SELECT [^;]+;$/.test(line))) throw Error('Read-only transaction required')
  return sql
}
export function buildCompletionEvidenceSql(scope) {
  identifier(scope?.customerId); identifier(scope?.rootOrderId)
  if (scope.eventType !== 'FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' || !Array.isArray(scope.sourceOrderIds)
    || scope.sourceOrderIds.length !== 2 || new Set(scope.sourceOrderIds).size !== 2
    || !scope.sourceOrderIds.includes(scope.rootOrderId)) throw Error('Invalid completion evidence scope')
  scope.sourceOrderIds.forEach(identifier)
  return [...transactionStart,
    `SELECT JSON_OBJECT('orderId',CAST(o.id AS CHAR),'customerId',CAST(o.tenant_id AS CHAR),'rootOrderId',CAST(o.root_order_id AS CHAR),'eventType','FIRST_LEG_ORDER_FULFILLMENT_COMPLETED','count',COUNT(x.id),'publishedCount',COALESCE(SUM(x.publish_status='PUBLISHED'),0),'pendingCount',COALESCE(SUM(x.publish_status='PENDING'),0),'failedCount',COALESCE(SUM(x.publish_status NOT IN ('PENDING','PUBLISHED')),0)) FROM ord_order o LEFT JOIN ord_outbox_event x ON x.aggregate_id=o.id AND x.tenant_id=o.tenant_id AND x.event_type='FIRST_LEG_ORDER_FULFILLMENT_COMPLETED' WHERE o.tenant_id=${scope.customerId} AND o.root_order_id=${scope.rootOrderId} AND o.id IN (${scope.sourceOrderIds.join(',')}) GROUP BY o.id,o.tenant_id,o.root_order_id ORDER BY o.id;`,
    'COMMIT;'].join('\n')
}
export function completionEvidenceFromRows(scope, rows) {
  buildCompletionEvidenceSql(scope)
  if (!Array.isArray(rows) || rows.length !== 2 || new Set(rows.map(row => row.orderId)).size !== 2
    || rows.some(row => !scope.sourceOrderIds.includes(row.orderId) || row.customerId !== scope.customerId
      || row.rootOrderId !== scope.rootOrderId || row.eventType !== scope.eventType
      || ['count','publishedCount','pendingCount','failedCount'].some(key => !Number.isSafeInteger(row[key]) || row[key] < 0)))
    throw Error('Completion evidence scope or counts mismatch')
  return {readOnly:true,customerId:scope.customerId,rootOrderId:scope.rootOrderId,orders:rows}
}
export const EVIDENCE_ENV_KEYS = ['JENKINS_URL','JENKINS_USER','JENKINS_API_TOKEN','FBA_CORE_JENKINS_AGENT']
export function validateEvidenceConfig(env) {
  if (EVIDENCE_ENV_KEYS.some(key => typeof env[key] !== 'string' || !env[key].trim())) throw Error('Missing TEST evidence configuration')
  const url = new URL(env.JENKINS_URL)
  if (!['http:','https:'].includes(url.protocol) || url.username || url.password || url.search || url.hash
    || !/^[A-Za-z0-9_.-]{1,128}$/.test(env.FBA_CORE_JENKINS_AGENT)) throw Error('Invalid TEST evidence configuration')
}
const base64Expression = value => '['+Buffer.from(value).toString('base64').match(/.{1,4000}/g).map(part=>`'${part}'`).join(',')+"].join('')"
// Credentials stay in the executor's memory. No deploy, SQL mutation, or arbitrary SQL entry point.
export function buildEvidenceRemoteScript(service, sql, agent) {
  assertReadOnlySql(sql)
  if (!['order','firstleg','lastleg'].includes(service) || !/^[A-Za-z0-9_.-]{1,128}$/.test(agent)) throw Error('Invalid TEST evidence target')
  const prefix=service.toUpperCase(),container=service==='firstleg'?'firstleg-test-business':`firstleg-test-${service}`
  const python = `import json,os,subprocess,urllib.parse,base64,re
try:
 c=json.loads(subprocess.check_output(['docker','inspect','${container}'],text=True,stderr=subprocess.DEVNULL))[0]
 if c.get('Name')!='/${container}' or c.get('Config',{}).get('Labels',{}).get('com.docker.compose.project')!='firstleg-test': raise RuntimeError('target')
 e=dict(v.split('=',1) for v in c['Config']['Env'])
 u=urllib.parse.urlsplit(e['${prefix}_DB_URL'].removeprefix('jdbc:'))
 if not u.hostname or not u.path.lstrip('/'): raise RuntimeError('database')
 childEnv=dict(os.environ,MYSQL_PWD=e['${prefix}_DB_PASSWORD'])
 host='127.0.0.1' if u.hostname=='host.docker.internal' else u.hostname
 sql=base64.b64decode('${Buffer.from(sql).toString('base64')}').decode('utf-8')
 p=subprocess.run(['mysql','--host='+host,'--port='+str(u.port or 3306),'--user='+e['${prefix}_DB_USERNAME'],'--batch','--raw','--skip-column-names',u.path.lstrip('/')],input=sql,capture_output=True,text=True,env=childEnv,timeout=40)
 if p.returncode:
  code=re.search(r'ERROR ([0-9]{3,5})',p.stderr)
  print('FBA_CORE_JSON='+json.dumps({'ok':False,'error':'MYSQL_'+(code.group(1) if code else 'UNKNOWN')}))
 else: print('FBA_CORE_JSON='+json.dumps({'ok':True,'rows':[json.loads(line) for line in p.stdout.splitlines() if line.strip()]}))
except Exception:
 print('FBA_CORE_JSON='+json.dumps({'ok':False,'error':'REMOTE_QUERY_FAILED'}))
`
  const groovy=`def p=new ProcessBuilder(['python3','-c',new String(java.util.Base64.decoder.decode(${base64Expression(python)}),'UTF-8')]).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 evidence agent unavailable'); print(hudson.util.RemotingDiagnostics.executeGroovy(new String(java.util.Base64.decoder.decode(${base64Expression(groovy)}),'UTF-8'),c.channel));`
}
export function createTestEvidenceReaders(env, fetcher = globalThis.fetch) {
  validateEvidenceConfig(env)
  async function execute(service, sql) {
    const script=buildEvidenceRemoteScript(service,sql,env.FBA_CORE_JENKINS_AGENT)
    let response
    try {
      response=await fetcher(new URL('scriptText',env.JENKINS_URL.replace(/\/?$/,'/')),{
        method:'POST',headers:{Authorization:'Basic '+Buffer.from(env.JENKINS_USER+':'+env.JENKINS_API_TOKEN).toString('base64'),
          'Content-Type':'application/x-www-form-urlencoded'},body:new URLSearchParams({script}),signal:AbortSignal.timeout(55000)})
    } catch { throw Error('TEST evidence transport failed') }
    if (!response.ok) throw Error('TEST evidence HTTP '+response.status)
    let raw
    try { raw=await response.text() } catch { throw Error('TEST evidence response failed') }
    const lines=raw.split(/\r?\n/).filter(line=>line.startsWith('FBA_CORE_JSON='))
    if(lines.length!==1) throw Error('TEST evidence JSON missing or ambiguous')
    let result
    try { result=JSON.parse(lines[0].slice('FBA_CORE_JSON='.length)) } catch { throw Error('TEST evidence JSON invalid') }
    if(result?.ok!==true || !Array.isArray(result.rows)) throw Error(/^MYSQL_[0-9]{3,5}$/.test(result?.error)?result.error:'TEST evidence query failed')
    return result.rows
  }
  return {
    readFleetInventory: async scope => fleetInventoryFromRows(scope, await execute('lastleg', buildFleetInventorySql(scope))),
    readReceiving: createReceivingEvidenceReader(sql => execute('lastleg',sql)),
    readCompletion: async scope => completionEvidenceFromRows(scope,await execute('order',buildCompletionEvidenceSql(scope))),
    preflight: async () => {
      for(const service of ['order','lastleg']) {
        const rows=await execute(service,[...transactionStart,"SELECT JSON_OBJECT('testProbe',1);",'COMMIT;'].join('\n'))
        if(rows.length!==1 || rows[0].testProbe!==1)throw Error('TEST evidence preflight failed')
      }
    },
  }
}