188 lines
15 KiB
JavaScript
188 lines
15 KiB
JavaScript
// 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=`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')
|
|
}
|
|
},
|
|
}
|
|
}
|