// Requires Node.js 24 or newer. Uses Node.js built-ins only and makes no network requests.
import {mkdtemp,readFile,rm,writeFile} from 'node:fs/promises'
import {spawnSync} from 'node:child_process'
import {tmpdir} from 'node:os'
import path from 'node:path'
import {fileURLToPath} from 'node:url'

const own=fileURLToPath(import.meta.url)
const exact=(value,keys,label)=>{if(!value||typeof value!=='object'||Array.isArray(value))throw new Error(`${label} must be an object`);for(const key of Object.keys(value))if(!keys.includes(key))throw new Error(`unknown ${label} field: ${key}`)}
const text=value=>typeof value==='string'&&value.trim().length>0
const safeId=value=>text(value)&&value.length<=64&&/^[a-z0-9][a-z0-9._-]*$/.test(value)&&!['__proto__','constructor','prototype','tostring'].includes(value.toLowerCase())
const ownKey=(object,key)=>Object.prototype.hasOwnProperty.call(object,key)
const readJson=async file=>JSON.parse(await readFile(file,'utf8'))
const saveJson=(file,value)=>writeFile(file,`${JSON.stringify(value,null,2)}\n`)

export function validateScenario(value){
  exact(value,['schema','maxAttempts','crashAfterSinkAccept','sinkFailuresBeforeAccept','events'],'scenario')
  if(value.schema!=='martybytes-notification-lab/v1')throw new Error('unsupported scenario schema')
  if(!Number.isInteger(value.maxAttempts)||value.maxAttempts<1||value.maxAttempts>9)throw new Error('maxAttempts must be an integer from 1 to 9')
  if(!text(value.crashAfterSinkAccept))throw new Error('crashAfterSinkAccept must be a non-blank string')
  exact(value.sinkFailuresBeforeAccept,Object.keys(value.sinkFailuresBeforeAccept??{}),'sinkFailuresBeforeAccept')
  for(const [id,count] of Object.entries(value.sinkFailuresBeforeAccept))if(!text(id)||!Number.isInteger(count)||count<0)throw new Error('sink failure counts must be non-negative integers')
  if(!Array.isArray(value.events)||!value.events.length||value.events.length>100)throw new Error('events must contain 1 to 100 records')
  const eventIds=new Set()
  for(const [index,event] of value.events.entries()){
    exact(event,['sequence','eventId','taskId','kind','message'],`event ${index}`)
    if(!Number.isInteger(event.sequence)||event.sequence<1)throw new Error(`event ${index} sequence must be a positive integer`)
    if(!safeId(event.eventId)||!safeId(event.taskId))throw new Error(`event ${index} identifiers must use 1 to 64 lowercase letters, numbers, dots, underscores, or hyphens`)
    if(!text(event.message)||event.message.length>200)throw new Error(`event ${index} message must contain 1 to 200 characters`)
    if(!['completed','failed','needs_attention'].includes(event.kind))throw new Error(`event ${index} kind is invalid`)
    eventIds.add(event.eventId)
  }
  if(!value.events.some(event=>event.eventId===value.crashAfterSinkAccept))throw new Error('crash event must exist')
  for(const id of Object.keys(value.sinkFailuresBeforeAccept))if(!eventIds.has(id))throw new Error(`sink failure references unknown event: ${id}`)
  return value
}

const initialState=()=>({schema:'martybytes-notification-state/v1',events:{},inputDuplicates:0,outOfOrderObserved:false,lastInputSequence:null,deadLetters:[]})
const initialSink=()=>({schema:'martybytes-fake-sink/v1',attempts:{},accepted:{}})
const loadOr=async(file,fallback)=>{try{return await readJson(file)}catch(error){if(error.code==='ENOENT')return fallback();throw error}}

async function ingest(scenario,stateFile){
  const state=await loadOr(stateFile,initialState)
  for(const event of scenario.events){
    if(state.lastInputSequence!==null&&event.sequence<state.lastInputSequence)state.outOfOrderObserved=true
    state.lastInputSequence=event.sequence
    const prior=ownKey(state.events,event.eventId)?state.events[event.eventId]:null
    if(prior){
      if(JSON.stringify(prior.event)!==JSON.stringify(event))throw new Error(`conflicting duplicate event: ${event.eventId}`)
      state.inputDuplicates++
      continue
    }
    state.events[event.eventId]={event,status:'pending',dispatchAttempts:0,lastError:null,errorHistory:[],ack:null}
  }
  await saveJson(stateFile,state)
}

async function sendToFakeSink(scenario,event,sinkFile){
  const sink=await loadOr(sinkFile,initialSink)
  sink.attempts[event.eventId]=(ownKey(sink.attempts,event.eventId)?sink.attempts[event.eventId]:0)+1
  const attempt=sink.attempts[event.eventId]
  if(attempt<=(scenario.sinkFailuresBeforeAccept[event.eventId]??0)){
    await saveJson(sinkFile,sink)
    throw new Error(`synthetic sink failure ${attempt}`)
  }
  if(ownKey(sink.accepted,event.eventId)){await saveJson(sinkFile,sink);return{ack:'duplicate',sinkAttempt:attempt}}
  sink.accepted[event.eventId]={sequence:event.sequence,taskId:event.taskId,kind:event.kind,message:event.message}
  await saveJson(sinkFile,sink)
  return{ack:'accepted',sinkAttempt:attempt}
}

async function dispatch(scenario,stateFile,sinkFile,{injectCrash=false}={}){
  const state=await readJson(stateFile)
  const ordered=Object.values(state.events).sort((a,b)=>a.event.sequence-b.event.sequence||a.event.eventId.localeCompare(b.event.eventId))
  for(const record of ordered){
    if(record.status!=='pending')continue
    while(record.errorHistory.length<scenario.maxAttempts){
      record.dispatchAttempts++
      await saveJson(stateFile,state)
      try{
        const result=await sendToFakeSink(scenario,record.event,sinkFile)
        if(injectCrash&&record.event.eventId===scenario.crashAfterSinkAccept&&result.ack==='accepted')process.exit(75)
        record.status='acknowledged';record.ack=result.ack;record.lastError=null
        await saveJson(stateFile,state)
        break
      }catch(error){
        record.lastError=error.message
        record.errorHistory.push({attempt:record.dispatchAttempts,error:record.lastError})
        await saveJson(stateFile,state)
        if(record.errorHistory.length===scenario.maxAttempts){
          record.status='dead-letter'
          state.deadLetters.push({eventId:record.event.eventId,attempts:record.dispatchAttempts,lastError:record.lastError})
          await saveJson(stateFile,state)
        }
      }
    }
  }
}

async function summarize(stateFile,sinkFile,phaseExits){
  const state=await readJson(stateFile),sink=await readJson(sinkFile)
  const records=Object.values(state.events).sort((a,b)=>a.event.sequence-b.event.sequence||a.event.eventId.localeCompare(b.event.eventId))
  return {
    schema:'martybytes-notification-result/v1',
    phaseExits,
    input:{records:records.length,duplicates:state.inputDuplicates,outOfOrderObserved:state.outOfOrderObserved},
    delivery:{acknowledged:records.filter(x=>x.status==='acknowledged').length,deadLetter:records.filter(x=>x.status==='dead-letter').length,sinkAccepted:Object.keys(sink.accepted).length},
    events:records.map(({event,status,dispatchAttempts,lastError,errorHistory,ack})=>({eventId:event.eventId,sequence:event.sequence,kind:event.kind,status,dispatchAttempts,sinkAttempts:sink.attempts[event.eventId]??0,ack,lastError,errorHistory})),
    claims:{deliverySemantics:'at-least-once attempts with sink idempotency by stable eventId',exactlyOnce:false,humanNotified:false}
  }
}

export async function runLab(scenarioFile,outputFile){
  const scenario=validateScenario(await readJson(scenarioFile)),dir=await mkdtemp(path.join(tmpdir(),'martybytes-notification-lab-'))
  try{
    const stateFile=path.join(dir,'state.json'),sinkFile=path.join(dir,'sink.json')
    await ingest(scenario,stateFile)
    const childOptions={encoding:'utf8',timeout:10000,maxBuffer:1024*1024}
    const phase1=spawnSync(process.execPath,[own,'dispatch',scenarioFile,stateFile,sinkFile,'--crash'],childOptions)
    const phase2=spawnSync(process.execPath,[own,'dispatch',scenarioFile,stateFile,sinkFile],childOptions)
    const result=await summarize(stateFile,sinkFile,[phase1.status,phase2.status])
    if(phase1.status!==75||phase2.status!==0)throw new Error(`unexpected phase exits: ${phase1.status},${phase2.status}`)
    if(result.events.some(event=>event.status==='pending'))throw new Error('restart left an event pending')
    await writeFile(outputFile,`${JSON.stringify(result,null,2)}\n`,{flag:'wx'})
    return result
  }finally{await rm(dir,{recursive:true,force:true})}
}

const [mode,...args]=process.argv.slice(2)
try{
  if(mode==='dispatch'){
    const [scenarioFile,stateFile,sinkFile,flag]=args
    await dispatch(validateScenario(await readJson(scenarioFile)),stateFile,sinkFile,{injectCrash:flag==='--crash'})
  }else if(mode==='lab'){
    if(args.length!==2)throw new Error('usage: notification-delivery-lab.mjs lab SCENARIO OUTPUT')
    await runLab(path.resolve(args[0]),path.resolve(args[1]))
  }else if(process.argv[1]&&path.resolve(process.argv[1])===own)throw new Error('mode must be lab or dispatch')
}catch(error){console.error(error.message);process.exitCode=1}