// 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') } }, } }