import test from 'node:test'
import assert from 'node:assert/strict'
import { spawnSync } from 'node:child_process'
import { validateRacePlan,buildOrderLockLeaseScript,buildLockEvidenceSql,buildStockEvidenceSql,assertContendedRequests,assertRaceInventory } from './f32-pda-hold-race.mjs'
import {assertReadOnlySql} from './fba-core-regression-evidence.mjs'
const plan={purpose:'F32_OWNED_PDA_HOLD_RACE',marker:'OFFLINE',customerId:'21',orderId:'22',warehouseId:'23',balanceId:'24',positionId:'25',pdaMembershipId:'26',dispatchItemId:'27',dispatchNo:'DS-OFFLINE',tripNo:'TR-OFFLINE',expectedImage:'sha256:'+'a'.repeat(64),expectedPackages:4,pickPackages:1,pickQuantity:10}
test('explicit owned scope rejects SQL injection, image drift and widened quantities',()=>{
  assert.equal(validateRacePlan(plan),plan)
  for(const changed of [{password:'do not persist'},{orderId:'22;DELETE FROM ll_order'},{customerId:21},{positionId:'0'},{expectedImage:'latest'},{marker:'x\';evil'},{expectedPackages:5},{pickPackages:2},{pickQuantity:0}])
    assert.throws(()=>validateRacePlan({...plan,...changed}))
  assert.throws(()=>buildOrderLockLeaseScript(plan,'OTHER','agent'))
  assert.throws(()=>buildOrderLockLeaseScript(plan,'HOLD_FIRST',"agent');evil"))
})
test('both evidence queries remain consistent read-only transactions with exact original source filters',()=>{
  for(const phase of ['HOLD_FIRST','PICK_FIRST']){
    const sql=buildLockEvidenceSql(plan,phase)
    assert.equal(assertReadOnlySql(sql),sql)
    assert.match(sql,/OBJECT_NAME='ll_order'/);assert.match(sql,/INDEX_NAME='PRIMARY'/)
    assert.match(sql,/LOCK_DATA=CONVERT\(0x3232/)
  }
  const sql=buildStockEvidenceSql(plan)
  assert.equal(assertReadOnlySql(sql),sql)
  assert.match(sql,/dispatch_item_id=27/);assert.match(sql,/order_id=22/);assert.match(sql,/balance_id=24/)
})
function chunks(source){return Array.from(source.matchAll(/\[('[A-Za-z0-9+/=]+'(?:,'[A-Za-z0-9+/=]+')*)\]\.join\(''\)/g),m=>Array.from(m[1].matchAll(/'([^']+)'/g),x=>x[1]).join(''))}
test('lease has a fixed bounded server sleep, rollback and immutable healthy TEST target; Python compiles offline',()=>{
  const outer=buildOrderLockLeaseScript(plan,'HOLD_FIRST','agent')
  const inner=Buffer.from(chunks(outer)[0],'base64').toString('utf8')
  const python=Buffer.from(chunks(inner)[0],'base64').toString('utf8')
  assert.match(python,/SLEEP\(12\)/);assert.match(python,/ROLLBACK;/);assert.match(python,/timeout=22/)
  assert.ok(python.includes("JSON_OBJECT('marker','F32_OFFLINE_HOLD_FIRST'"),'Correlation must survive the MySQL client stripping comments')
  assert.match(python,/JSON_EXTRACT\('true','\$'\)/)
  assert.match(python,/innodb_lock_wait_timeout=5/);assert.match(python,/FOR UPDATE/)
  assert.match(python,/DEPLOY_ENV.*test/);assert.match(python,/MYSQL_PWD=e\['LASTLEG_DB_PASSWORD'\]/)
  assert.ok(python.includes(plan.expectedImage));assert.match(python,/Health.*healthy/)
  assert.doesNotMatch(python,/INSERT INTO|DELETE FROM|UPDATE ll_order|docker.*(?:stop|restart)/)
  const result=spawnSync(process.env.PYTHON||'python',['-c',"import sys;compile(sys.stdin.read(),'<offline-lease>','exec')"],{input:python,encoding:'utf8'})
  assert.equal(result.status,0,result.stderr||result.error?.message)
})
const rows=[{collection:'lease',connectionId:100},{collection:'wait',requestingConnectionId:101,blockingConnectionId:100,orderId:'22'}]
function scanSnapshots(){
  const bucket=available=>({availablePackages:available,holdPackages:0,allocatedPackages:0,pickedPackages:0,receivingPackages:0})
  const before={balance:{...bucket(4),totalPackages:4},positions:[{id:'25',...bucket(3)},{id:'28',...bucket(1)}],detail:{order:{detained:false}},sql:[]}
  const after={balance:{...bucket(3),totalPackages:3},positions:[{id:'25',...bucket(2)},{id:'28',...bucket(1)}],detail:{order:{detained:false,version:1}},sql:[
    {collection:'sourceOrder',id:'22',customerId:'21',warehouseId:'23',picked:1,shipped:0,detained:0,version:1},
    {collection:'pick',id:'100',key:'OFFLINE-PICK_FIRST-PICK',positionId:'25',packages:1,quantity:10},
    {collection:'attribution',id:'101',pickId:'100',orderId:'22',packages:1,quantity:10},
    ...[['ALLOCATE','AVAILABLE','ALLOCATED',1],['PICK','ALLOCATED','PICKED',1],['SHIP','PICKED','SHIPPED',-1]].map(([businessType,statusFrom,statusTo,delta],i)=>({collection:'ledger',id:String(102+i),key:`DISPATCH:DS-OFFLINE:27:OFFLINE-PICK_FIRST-PICK:${businessType}:operation`,positionId:'25',businessType,statusFrom,statusTo,delta,before:delta>0?0:1,after:delta>0?1:0})),
  ]}
  return{before,after}
}
test('successful PDA scan accounts for one consumed box and its three immutable ledger steps, without inventing DTO counters',()=>{
  const {before,after}=scanSnapshots()
  assert.doesNotThrow(()=>assertRaceInventory(plan,'PICK_FIRST',before,after))
})
test('missing/duplicate scan ledger, wrong source and a second deduction cannot pass the inventory proof',()=>{
  for(const change of [x=>x.sql=x.sql.filter(y=>y.businessType!=='SHIP'),x=>x.sql.push({...x.sql.find(y=>y.businessType==='SHIP'),id:'999'}),
    x=>x.sql.find(y=>y.collection==='sourceOrder').picked=2,x=>x.sql.find(y=>y.collection==='attribution').orderId='other',
    x=>x.balance.availablePackages=2,x=>x.sql.find(y=>y.businessType==='SHIP').positionId='other',x=>x.sql.find(y=>y.businessType==='SHIP').delta=-2]){
    const {before,after}=scanSnapshots();change(after)
    assert.throws(()=>assertRaceInventory(plan,'PICK_FIRST',before,after))
  }
})
test('hold winner moves the exact two positions, with one history and no pick or attribution facts',()=>{
  const {before}=scanSnapshots()
  const after={balance:{...before.balance,availablePackages:0,holdPackages:4},positions:before.positions.map(x=>({...x,availablePackages:0,holdPackages:x.availablePackages})),
    detail:{order:{detained:true,version:2}},sql:[{collection:'sourceOrder',id:'22',customerId:'21',warehouseId:'23',picked:0,shipped:0,detained:1,version:2},
      {collection:'hold',id:'200',key:'OFFLINE-HOLD_FIRST-HOLD',held:1,version:2},
      ...before.positions.map((x,i)=>({collection:'ledger',id:String(201+i),key:'hold',positionId:x.id,businessType:'HOLD',statusFrom:'AVAILABLE',statusTo:'HOLD',before:0,delta:x.availablePackages,after:x.availablePackages}))]}
  assert.equal(assertRaceInventory(plan,'HOLD_FIRST',before,after).heldPackages,4)
  for(const change of [x=>x.sql.pop(),x=>x.sql.push({collection:'pick',id:'300',key:'wrong'}),x=>x.sql.find(x=>x.collection==='hold').version=3,x=>x.sql.find(x=>x.collection==='ledger').positionId='other']){
    const changed=structuredClone(after);change(changed)
    assert.throws(()=>assertRaceInventory(plan,'HOLD_FIRST',before,changed))
  }
})
test('concurrency requires a live original row lease, first observed waiter and two distinct simultaneous waiters',()=>{
  assert.deepEqual(assertContendedRequests(rows,'22').requestingConnections,[101])
  const both=[...rows,{collection:'wait',requestingConnectionId:102,blockingConnectionId:100,orderId:'22'}]
  assert.deepEqual(assertContendedRequests(both,'22',101).requestingConnections,[101,102])
  assert.throws(()=>assertContendedRequests(rows,'22',101),/EXACT_REQUEST_CONTENTION/)
  assert.throws(()=>assertContendedRequests(both,'22',999),/FIRST_REQUEST_NO_LONGER_WAITING/)
  assert.throws(()=>assertContendedRequests(both,'other',101),/LOCK_EVIDENCE_SCOPE/)
  assert.throws(()=>assertContendedRequests(rows.slice(1),'22'),/LIVE_LEASE/)
  assert.throws(()=>assertContendedRequests([...both,{collection:'wait',requestingConnectionId:103,blockingConnectionId:100,orderId:'22'}],'22',101),/EXACT_REQUEST_CONTENTION/)
})