import assert from 'node:assert/strict'
import { spawn, spawnSync } from 'node:child_process'
import { createServer } from 'node:http'
import { mkdtemp, readFile, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join, dirname, basename, resolve } from 'node:path'
import { fileURLToPath } from 'node:url'
import test from 'node:test'

const script = fileURLToPath(new URL('./event-delivery-fault-proxy.py', import.meta.url))
const targetId = 'e6259b1a-e112-4233-83e1-ae0071db3c52'
const otherId = '1b344fb7-d5d9-4324-b2d1-51bd0ab3af55'
const samePlanOtherId = '05dab8d1-10fd-4c9e-b242-54f4c6ff76a4'
const hash = `sha256:${'a'.repeat(64)}`
const secret = 'test-service-identity-secret-do-not-log'
const controlToken = 'test-control-secret-with-more-than-24-characters'
const event = (id = targetId, planId = '2100000000000000001') => ({ eventId: id,
  eventType: 'LoadingCompleted', payloadHash: hash,
  payload: { loadingPlanId: planId, syntheticPrivateField: secret } })
const close = server => new Promise(resolve => server.close(resolve))

async function fixture(selector = ['--loading-plan-id', '2100000000000000001'], eventType = 'LoadingCompleted', endpoint = '/internal/events') {
  const calls = []
  const applied = new Set()
  let invalidAck = false
  const consumer = createServer(async (request, response) => {
    const chunks = []
    for await (const chunk of request) chunks.push(chunk)
    const raw = Buffer.concat(chunks).toString()
    const envelope = JSON.parse(raw)
    calls.push({ raw, authorization: request.headers.authorization, eventId: envelope.eventId, path: request.url })
    const outcome = applied.has(envelope.eventId) ? 'DUPLICATE' : 'APPLIED'
    applied.add(envelope.eventId)
    const body = JSON.stringify({ code: '200', data: { eventId: invalidAck ? otherId : envelope.eventId,
      payloadHash: envelope.payloadHash, outcome } })
    response.writeHead(200, { 'Content-Type': 'application/json', 'X-Consumer-Evidence': 'actual-upstream' })
    response.end(body)
  })
  await new Promise(resolve => consumer.listen(0, '127.0.0.1', resolve))
  const directory = await mkdtemp(join(tmpdir(), 'event-fault-offline-'))
  const log = join(directory, 'evidence.jsonl')
  const child = spawn(process.env.PYTHON || 'python', [script, '--upstream',
    `http://127.0.0.1:${consumer.address().port}${endpoint}`, '--event-type', eventType,
    ...selector, '--log', log], { env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken },
    stdio: ['ignore', 'pipe', 'pipe'] })
  let output = ''; let errors = ''
  child.stdout.on('data', chunk => { output += chunk })
  child.stderr.on('data', chunk => { errors += chunk })
  // Python 启动失败时也必须关闭本地消费者，否则失败用例会占住整个测试进程。
  let ready
  try { ready = await new Promise((resolve, reject) => {
    const timer = setTimeout(() => { child.kill(); reject(new Error('Proxy readiness timeout')) }, 8000)
    const onData = () => {
      const line = output.split('\n').find(item => item.trim())
      if (!line) return
      try { const value = JSON.parse(line); clearTimeout(timer); child.stdout.off('data', onData); resolve(value) }
      catch { clearTimeout(timer); reject(new Error('Invalid proxy readiness JSON')) }
    }
    child.stdout.on('data', onData)
    child.once('error', error => { clearTimeout(timer); reject(error) })
    child.once('exit', code => { if (!output) { clearTimeout(timer); reject(new Error(`Proxy exit ${code}`)) } })
  }) } catch (error) {
    if (child.exitCode === null && !child.killed) child.kill()
    await close(consumer)
    assert.equal(dirname(resolve(directory)), resolve(tmpdir()))
    assert.ok(basename(directory).startsWith('event-fault-offline-'))
    await rm(directory, { recursive: true })
    throw error
  }
  const dataUrl = `http://127.0.0.1:${ready.dataPort}${endpoint}`
  const controlUrl = `http://127.0.0.1:${ready.controlPort}`
  return { calls, applied, log, invalidAck: value => { invalidAck = value },
    send: async (envelope, path = endpoint) => {
      const sent = JSON.stringify(envelope)
      const response = await fetch(new URL(path, dataUrl), { method: 'POST', body: sent,
        headers: { 'Content-Type': 'application/json', Authorization: `Bearer ${secret}` } })
      return { status: response.status, raw: await response.text(), headers: response.headers, sent }
    },
    control: (path, body, token = controlToken) => fetch(`${controlUrl}${path}`, {
      method: body ? 'POST' : 'GET', headers: { 'X-Event-Fault-Control-Token': token,
        'Content-Type': 'application/json' }, body: body ? JSON.stringify(body) : undefined,
    }),
    output: () => `${output}\n${errors}`,
    stop: async () => {
      if (child.exitCode === null) {
        const exited = new Promise(resolve => child.once('exit', resolve))
        child.kill(); await exited
      }
      await close(consumer)
      assert.equal(dirname(resolve(directory)), resolve(tmpdir()))
      assert.ok(basename(directory).startsWith('event-fault-offline-'))
      await rm(directory, { recursive: true })
    },
  }
}

test('fee fault preserves wallet endpoint bytes and auth without selecting its body', async t => {
  const proxy = await fixture(['--fee-source-id', '2100000000000000001', '--source-version', '1'],
    'BusinessFeeSourceChanged', '/internal/finance/v1/business-fee-source-events')
  t.after(() => proxy.stop())
  await proxy.control('/mode', { mode: 'FAULT' })
  const envelope = { ...event(), eventType: 'BusinessFeeSourceChanged',
    payload: { sourceFeeId: '2100000000000000001', sourceVersion: 1, action: 'VOIDED' } }
  const walletPath = '/internal/finance/v1/wallet-command-events'
  const wallet = await proxy.send(envelope, walletPath)
  assert.equal(wallet.status, 200)
  assert.equal(proxy.calls[0].path, walletPath)
  assert.equal(proxy.calls[0].raw, wallet.sent)
  assert.equal(proxy.calls[0].authorization, `Bearer ${secret}`)
  assert.equal((await (await proxy.control('/status')).json()).attempt, 0)
  assert.equal((await proxy.send(envelope, '/internal/finance/v1/unknown')).status, 404)
  assert.equal(proxy.calls.length, 1)
  assert.equal((await proxy.send(envelope)).status, 503)
  assert.equal((await (await proxy.control('/status')).json()).attempt, 1)
})

test('ACK-loss proxy only targets one event and forwards committed upstream results before dropping ACK', async t => {
  const proxy = await fixture()
  t.after(() => proxy.stop())
  const unauthorized = await proxy.control('/mode', { mode: 'FAULT' }, 'wrong-control-token')
  assert.equal(unauthorized.status, 403)
  let status = await (await proxy.control('/status')).json()
  assert.equal(status.mode, 'PASS')
  assert.equal((await proxy.control('/mode', { mode: 'FAULT' })).status, 200)

  await t.test('real consumer is reached and durable APPLIED then DUPLICATE both lose ACK', async () => {
    const first = await proxy.send(event())
    assert.equal(first.status, 503)
    assert.equal(JSON.parse(first.raw).errorCode, 'ACK_LOSS_INJECTED')
    assert.ok(proxy.applied.has(targetId))
    const duplicate = await proxy.send(event())
    assert.equal(duplicate.status, 503)
    assert.equal(proxy.calls.length, 2)
    assert.equal(proxy.calls[0].raw, first.sent)
    assert.equal(proxy.calls[0].authorization, `Bearer ${secret}`)
    status = await (await proxy.control('/status')).json()
    assert.deepEqual(status.records.map(item => item.outcome), ['APPLIED', 'DUPLICATE'])
    assert.deepEqual(status.records.map(item => item.attempt), [1, 2])
  })

  await t.test('other plans, other types and another same-plan ID are unaffected', async () => {
    for (const envelope of [event(otherId, '2100000000000000002'),
      { ...event(otherId), eventType: 'CarrierScheduleNodeUpserted' }, event(samePlanOtherId)]) {
      const result = await proxy.send(envelope)
      assert.equal(result.status, 200)
      assert.equal(JSON.parse(result.raw).data.eventId, envelope.eventId)
      assert.equal(result.headers.get('x-consumer-evidence'), 'actual-upstream')
    }
    status = await (await proxy.control('/status')).json()
    assert.equal(status.attempt, 2)
    assert.equal(status.records.length, 2)
  })

  await t.test('pinned identity rejects hash, hidden payload and malformed-hash changes before upstream', async () => {
    const before = proxy.calls.length
    for (const envelope of [{ ...event(), payloadHash: `sha256:${'b'.repeat(64)}` },
      { ...event(), payload: { loadingPlanId: '2100000000000000002', changed: secret } },
      { ...event(), payloadHash: secret }]) {
      const result = await proxy.send(envelope)
      assert.equal(result.status, 409)
      assert.equal(JSON.parse(result.raw).errorCode, 'TARGET_EVENT_HASH_CONFLICT')
    }
    assert.equal(proxy.calls.length, before)
  })

  await t.test('PASS forwards matching duplicate ACK and original upstream response bytes and headers', async () => {
    assert.equal((await proxy.control('/mode', { mode: 'PASS' })).status, 200)
    const result = await proxy.send(event())
    assert.equal(result.status, 200)
    assert.equal(result.raw, JSON.stringify({ code: '200', data: { eventId: targetId, payloadHash: hash, outcome: 'DUPLICATE' } }))
    assert.equal(result.headers.get('x-consumer-evidence'), 'actual-upstream')
    status = await (await proxy.control('/status')).json()
    assert.equal(status.records.at(-1).outcome, 'DUPLICATE')
    assert.equal(status.records.at(-1).errorCode, null)
  })

  await t.test('incorrect upstream ACK is never replaced with plausible success evidence', async () => {
    proxy.invalidAck(true)
    const result = await proxy.send(event())
    assert.equal(result.status, 502)
    assert.equal(JSON.parse(result.raw).errorCode, 'UPSTREAM_ACK_INVALID')
  })

  await t.test('state, JSONL evidence and subprocess output never expose private payload or identity secrets', async () => {
    const records = (await readFile(proxy.log, 'utf8')).trim().split('\n').map(line => JSON.parse(line))
    for (const record of records) {
      assert.deepEqual(Object.keys(record).sort(),
        ['time', 'eventId', 'payloadHash', 'outcome', 'attempt', 'matched', 'errorCode'].sort())
      assert.ok(record.eventId === targetId)
    }
    const evidence = `${JSON.stringify(await (await proxy.control('/status')).json())}\n${await readFile(proxy.log, 'utf8')}\n${proxy.output()}`
    assert.ok(!evidence.includes(secret))
    assert.ok(!evidence.includes(controlToken))
    assert.ok(!evidence.includes('syntheticPrivateField'))
    assert.ok(!evidence.includes('authorization'))
  })
})

test('explicit event-ID selector requires exact event type and forwards unrelated IDs unchanged', async t => {
  const proxy = await fixture(['--event-id', targetId])
  t.after(() => proxy.stop())
  await proxy.control('/mode', { mode: 'FAULT' })
  const other = await proxy.send(event(otherId))
  assert.equal(other.status, 200)
  const wrongType = await proxy.send({ ...event(), eventType: 'OverseasDeliveryCompleted' })
  assert.equal(wrongType.status, 200)
  const target = await proxy.send(event())
  assert.equal(target.status, 503)
  const status = await (await proxy.control('/status')).json()
  assert.equal(status.attempt, 1)
  assert.equal(status.eventId, targetId)
})

test('unloading selector pins only the exact source booking and preserves reverse-event ACK recovery', async t => {
  const bookingId = '2100000000000000066'
  const proxy = await fixture(['--container-booking-id', bookingId], 'OverseasContainerUnloaded')
  t.after(() => proxy.stop())
  await proxy.control('/mode', { mode: 'FAULT' })
  const unloaded = { ...event(), eventType: 'OverseasContainerUnloaded',
    payload: { sourceContainerBookingId: bookingId, containerNo: 'TEST-CONTAINER' } }
  for (const unrelated of [
    { ...unloaded, eventId: otherId, payload: { sourceContainerBookingId: '2100000000000000067' } },
    { ...unloaded, eventId: otherId, payload: { loadingPlanId: bookingId } },
    { ...unloaded, eventId: otherId, eventType: 'OverseasDeliveryCompleted' },
  ]) assert.equal((await proxy.send(unrelated)).status, 200)
  assert.equal((await proxy.send(unloaded)).status, 503)
  assert.equal((await proxy.send(unloaded)).status, 503)
  assert.equal((await proxy.send({ ...unloaded, eventId: samePlanOtherId })).status, 200)
  const before = proxy.calls.length
  assert.equal((await proxy.send({ ...unloaded, payload: { sourceContainerBookingId: '2100000000000000067' } })).status, 409)
  assert.equal(proxy.calls.length, before)
  await proxy.control('/mode', { mode: 'PASS' })
  const recovered = await proxy.send(unloaded)
  assert.equal(recovered.status, 200)
  assert.equal(JSON.parse(recovered.raw).data.outcome, 'DUPLICATE')
  const status = await (await proxy.control('/status')).json()
  assert.equal(status.eventId, targetId)
  assert.deepEqual(status.records.filter(row => row.errorCode !== 'TARGET_EVENT_HASH_CONFLICT').map(row => row.outcome),
    ['APPLIED', 'DUPLICATE', 'DUPLICATE'])
})

test('booking selector rejects malformed identities and mixed selection before starting listeners', () => {
  for (const args of [['--container-booking-id', '0'], ['--container-booking-id', '../66'],
    ['--container-booking-id', '66', '--loading-plan-id', '77'], ['--container-booking-id', '66', '--event-id', targetId]]) {
    const result = spawnSync(process.env.PYTHON || 'python', [script, '--upstream', 'http://127.0.0.1:1/internal/events',
      '--event-type', 'OverseasContainerUnloaded', ...args], { encoding: 'utf8', timeout: 5000,
      env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken } })
    assert.equal(result.status, 2)
    assert.ok(!result.stdout.includes('"ready"'))
  }
})

test('delivery selector isolates the first exact delivery event and preserves its ACK-loss recovery', async t => {
  const deliveryId = '2100000000000000088'
  const proxy = await fixture(['--delivery-id', deliveryId], 'OverseasDeliveryCompleted')
  t.after(() => proxy.stop())
  await proxy.control('/mode', { mode: 'FAULT' })
  const signed = { ...event(), eventType: 'OverseasDeliveryCompleted',
    payload: { deliveryOrderId: deliveryId, sourceInboundId: '2100000000000000099' } }
  for (const unrelated of [
    { ...signed, eventId: otherId, payload: { deliveryOrderId: '2100000000000000089' } },
    { ...signed, eventId: otherId, payload: { deliveryId } },
    { ...signed, eventId: otherId, payload: { loadingPlanId: deliveryId } },
    { ...signed, eventId: otherId, eventType: 'OverseasContainerUnloaded' },
  ]) assert.equal((await proxy.send(unrelated)).status, 200)
  assert.equal((await proxy.send(signed)).status, 503)
  assert.equal((await proxy.send(signed)).status, 503)
  assert.equal((await proxy.send({ ...signed, eventId: samePlanOtherId })).status, 200)
  const calls = proxy.calls.length
  assert.equal((await proxy.send({ ...signed, payload: { deliveryOrderId: '2100000000000000089' } })).status, 409)
  assert.equal(proxy.calls.length, calls)
  await proxy.control('/mode', { mode: 'PASS' })
  const recovered = await proxy.send(signed)
  assert.equal(recovered.status, 200)
  assert.equal(JSON.parse(recovered.raw).data.outcome, 'DUPLICATE')
  const status = await (await proxy.control('/status')).json()
  assert.equal(status.eventId, targetId)
  assert.deepEqual(status.records.filter(row => row.errorCode !== 'TARGET_EVENT_HASH_CONFLICT').map(row => row.outcome), ['APPLIED', 'DUPLICATE', 'DUPLICATE'])
  assert.ok(!(await readFile(proxy.log, 'utf8')).includes('sourceInboundId'))
})

test('delivery selection rejects malformed IDs or a second selector before opening a listener', () => {
  for (const args of [['--delivery-id', '0'], ['--delivery-id', '../88'], ['--delivery-id', '88', '--event-id', targetId],
    ['--delivery-id', '88', '--container-booking-id', '66'], ['--delivery-id', '88', '--loading-plan-id', '77']]) {
    const result = spawnSync(process.env.PYTHON || 'python', [script, '--upstream', 'http://127.0.0.1:1/internal/events',
      '--event-type', 'OverseasDeliveryCompleted', ...args], { encoding: 'utf8', timeout: 5000,
      env: { ...process.env, EVENT_FAULT_CONTROL_TOKEN: controlToken } })
    assert.equal(result.status, 2)
    assert.ok(!result.stdout.includes('"ready"'))
  }
})

test('fee void selector leaves publication, other sources and versions unchanged while recovering exact VOIDED ACK', async t => {
  const sourceFeeId='2100000000000000011'
  const proxy=await fixture(['--fee-source-id',sourceFeeId,'--source-version','2'],'BusinessFeeSourceChanged')
  t.after(()=>proxy.stop())
  await proxy.control('/mode',{mode:'FAULT'})
  const voided={...event(),eventType:'BusinessFeeSourceChanged',payload:{sourceFeeId,sourceVersion:2,action:'VOIDED',totalAmount:30}}
  for(const payload of [{...voided.payload,action:'PUBLISHED'},{...voided.payload,sourceVersion:1},{...voided.payload,sourceFeeId:'2100000000000000012'},
    {...voided.payload,action:'VOID'}]) assert.equal((await proxy.send({...voided,eventId:otherId,payload})).status,200)
  assert.equal((await proxy.send(voided)).status,503)
  assert.equal((await proxy.send(voided)).status,503)
  assert.equal((await proxy.send({...voided,eventId:samePlanOtherId})).status,200)
  const before=proxy.calls.length
  assert.equal((await proxy.send({...voided,payload:{...voided.payload,sourceVersion:3}})).status,409)
  assert.equal(proxy.calls.length,before)
  await proxy.control('/mode',{mode:'PASS'})
  assert.equal(JSON.parse((await proxy.send(voided)).raw).data.outcome,'DUPLICATE')
  const state=await (await proxy.control('/status')).json()
  assert.deepEqual(state.records.filter(r=>r.errorCode!=='TARGET_EVENT_HASH_CONFLICT').map(r=>r.outcome),['APPLIED','DUPLICATE','DUPLICATE'])
})

test('fee void selector requires a bounded version paired with only its own source selector',()=>{
  for(const args of [['--fee-source-id','11'],['--event-id',targetId,'--source-version','1'],['--fee-source-id','0','--source-version','1'],
    ['--fee-source-id','11','--source-version','0'],['--fee-source-id','11','--source-version','2147483648'],
    ['--fee-source-id','../11','--source-version','1'],['--fee-source-id','11','--source-version','1','--delivery-id','88']]) {
    const result=spawnSync(process.env.PYTHON||'python',[script,'--upstream','http://127.0.0.1:1/internal/events','--event-type','BusinessFeeSourceChanged',...args],
      {encoding:'utf8',timeout:5000,env:{...process.env,EVENT_FAULT_CONTROL_TOKEN:controlToken}})
    assert.equal(result.status,2);assert.ok(!result.stdout.includes('"ready"'))
  }
})