import assert from 'node:assert/strict'
import test from 'node:test'
import { readFileSync } from 'node:fs'
import { DISPATCH_BASE as BASE, verifyGroupedDispatch, groupedDispatchMatches, createBody, dispatchIo, sourceSelection, version, versionAfter } from './grouped-dispatch-acceptance.mjs'

const rows = () => [1, 2].map(n => ({ selection: { sourceType: 'INBOUND_PLAN', planPalletId: `2${n}`, returnReceiptId: null, relabelNo: null, warehousePalletId: `3${n}`, sourceVersion: `${n}`.repeat(64) },
  sourceKey: `PLAN:2${n}`, sourceVersion: `${n}`.repeat(64), customerId: '55', warehouseId: '44', inboundCode: `IN-${n}`, palletNo: `P-${n}`,
  destination: { key: `FBA:A${n}`, type: 'FBA', warehouseCode: `A${n}` }, availablePackages: n, appointmentBatches: [], blockReasons: [] }))
function server(fault = '') {
  const candidates = rows(), drafts = new Map(), orders = new Map(), ledger = new Map(), calls = [], checks = []
  let serial = 0, containerId = 60, itemId = 70
  const response = (data, code = '200') => ({ http: 200, body: { code, data: structuredClone(data) }, requestId: 'OFFLINE-REQ', traceId: 'OFFLINE-TRACE' })
  const context = { marker: 'CURRENT', createdResources: { phase4OverseasPda: { marker: 'CURRENT', finalStatus: 'COMPLETED', customerId: '55', overseasWarehouseId: '44', palletIds: ['31', '32'] } },
    checkpoint: async () => {}, // Explicit offline persistence double; production run supplies atomic disk persistence.
    assert: (_module, label, _response, passed) => checks.push({ label, passed }),
    async request(method, raw, body) {
      calls.push({ method, path: raw, body: structuredClone(body) })
      const path = raw.split('?')[0], params = new URLSearchParams(raw.split('?')[1]), write = method !== 'GET'
      const key = write ? `${method}:${path}:${body?.idempotencyKey}` : null
      if (write) {
        assert.ok(body?.idempotencyKey, raw)
        if (ledger.has(key)) { const old = ledger.get(key); if (old.body !== JSON.stringify(body)) return response(null, 'A0409'); const data = structuredClone(old.data); if (data?.vehicles) data.replayed = true; return response(data) }
      }
      const remember = data => { ledger.set(key, { body: JSON.stringify(body), data: structuredClone(data) }); return response(data) }
      const draft = body?.draftToken ? drafts.get(body.draftToken) : null
      if (path === `${BASE}/drafts` && method === 'POST') { assert.deepEqual(Object.keys(body).sort(), ['idempotencyKey', 'warehouseId']); const d = { draftToken: `SERVER-${++serial}`, recordVersion: 0, warehouseId: body.warehouseId, status: 'ACTIVE', containers: [], sources: [], expiresAt: '2099-01-01' }; drafts.set(d.draftToken, d); return remember(d) }
      if (path === `${BASE}/drafts` && method === 'GET') return response([...drafts.values()].filter(d => d.status === 'ACTIVE'))
      if (path === `${BASE}/candidates`) {
        assert.equal(params.get('warehouseId'), '44')
        const claimed = new Set([...orders.values()].flatMap(o => o.items.map(i => i.sourceKey)))
        return response({ complete: true, snapshotAt: '2026-10-19', candidates: candidates.filter(c => !claimed.has(c.sourceKey)) })
      }
      if (path === `${BASE}/drafts/containers`) { assert.equal(body.recordVersion, draft.recordVersion); assert.equal(body.capacity, 15); draft.containers.push({ draftContainerId: String(++containerId), sequenceNo: draft.containers.length + 1, capacity: 15, sourceKeys: [] }); draft.recordVersion++; return remember(draft) }
      if (path === `${BASE}/draft-locks/lock`) { assert.equal(body.recordVersion, draft.recordVersion); assert.equal(body.source.sourceVersion.length, 64); const c = candidates.find(c => c.selection.planPalletId === body.source.planPalletId); assert.ok(c); draft.sources.push(structuredClone(c)); draft.containers.find(g => g.draftContainerId === body.draftContainerId).sourceKeys.push(c.sourceKey); draft.recordVersion++; return remember(draft) }
      if (path === `${BASE}/drafts/snapshot`) { assert.equal(body.recordVersion, draft.recordVersion); assert.ok(Array.isArray(body.containers)); assert.equal(Object.hasOwn(body, 'containersJson'), false); draft.containers = structuredClone(body.containers); draft.recordVersion++; return remember(draft) }
      if (path === `${BASE}/drafts/pre-arrange`) {
        assert.equal(body.recordVersion, draft.recordVersion)
        const matches = body.groups.every(g => JSON.stringify(g.dispatch.cargo.map(s => `PLAN:${s.planPalletId}`).sort()) === JSON.stringify(draft.containers.find(c => c.draftContainerId === g.draftContainerId).sourceKeys.toSorted()))
        if (!matches) { if (fault === 'reject-mutates-draft') draft.containers[0].sourceKeys = []; return response(null, 'A0409') }
        const vehicles = body.groups.map(g => {
          assert.equal(g.dispatch.vehicle, null); assert.equal(g.dispatch.driver, null)
          const selected = candidates.filter(c => g.dispatch.cargo.some(s => s.planPalletId === c.selection.planPalletId))
          const number = `DS-${orders.size + 1}`, items = selected.map(c => ({ dispatchItemId: String(++itemId), sourceKey: c.sourceKey, sourceVersion: c.sourceVersion, customerId: c.customerId, planPalletId: c.selection.planPalletId, warehousePalletId: c.selection.warehousePalletId, inboundCode: c.inboundCode, destination: c.destination, scheduledPackages: c.availablePackages, pickedPackages: 0, remove: { allowed: true } }))
          const o = { dispatchNo: number, recordVersion: 1, items }; orders.set(number, o)
          return { draftContainerId: g.draftContainerId, dispatch: { dispatchNo: number, action: 'CREATE', status: 'ACTIVE', stage: 'BEFORE_LOADING', recordVersion: 1, replayed: false, counts: counts(o) } }
        })
        draft.status = 'CONVERTED'
        if (fault === 'unknown-convert') return { http: 0, body: null }
        if (fault === 'merged-groups') vehicles[1].dispatch.dispatchNo = vehicles[0].dispatch.dispatchNo
        return remember({ draftToken: draft.draftToken, vehicles, replayed: false })
      }
      const match = path.slice(BASE.length).match(/^\/([^/]+)(?:\/(items|route)(?:\/([^/]+)\/remove)?)?$/)
      if (path === BASE && method === 'GET') return response({ records: [...orders.values()].filter(o => o.dispatchNo === params.get('dispatchNo')).map(o => ({ dispatchNo: o.dispatchNo, warehouseId: '44', recordVersion: o.recordVersion, status: 'ACTIVE', stage: 'BEFORE_LOADING', counts: counts(o) })) })
      const o = orders.get(match?.[1]); assert.ok(o, `${method} ${raw}`)
      if (method === 'POST' && match[3]) { assert.equal(body.recordVersion, o.recordVersion); o.items = o.items.filter(i => i.dispatchItemId !== match[3]); o.recordVersion++; return remember({ dispatchNo: o.dispatchNo, action: 'REMOVE_CARGO', recordVersion: o.recordVersion, replayed: false }) }
      if (match[2] === 'items') return response({ dispatchNo: o.dispatchNo, recordVersion: o.recordVersion, status: 'ACTIVE', stage: 'BEFORE_LOADING', items: o.items })
      if (match[2] === 'route') return response({ dispatchNo: o.dispatchNo, recordVersion: o.recordVersion, stops: [{ index: 0, name: 'Origin', destinationKey: 'ORIGIN:44' }, ...o.items.map((i, n) => ({ index: n + 1, destinationKey: i.destination.key }))] })
      return response({ dispatchNo: o.dispatchNo, recordVersion: o.recordVersion, status: 'ACTIVE', stage: 'BEFORE_LOADING', counts: counts(o), dispatchInfo: { route: o.items.map(i => i.destination.warehouseCode).join(',') },
        orderDetails: o.items.map(i => ({ sourceKey: i.sourceKey, customerId: i.customerId, inboundCode: i.inboundCode, destinationFbaWarehouse: i.destination.warehouseCode })) })
    } }
  return { context, candidates, calls, checks, orders }
}
function counts(o) { return { itemCount: o.items.length, palletCount: o.items.length, packageCount: o.items.reduce((n, i) => n + i.scheduledPackages, 0), pickedPackageCount: 0, shippedPackageCount: 0 } }

test('actual controller and OpenAPI publish versioned server drafts; standalone legacy creation is absent', () => {
  const controller = readFileSync(new URL('../../lastleg-service/src/main/java/com/baiyue/lastleg/dispatch/controller/SmartDispatchController.java', import.meta.url), 'utf8')
  const yaml = readFileSync(new URL('../../lastleg-service/src/main/resources/openapi/overseas-business-v1.yaml', import.meta.url), 'utf8')
  for (const path of ['/drafts', '/drafts/containers', '/drafts/snapshot', '/draft-locks/lock', '/drafts/pre-arrange']) { assert.ok(controller.includes(`("${path}")`)); assert.ok(yaml.includes(`  ${BASE}${path}:`)) }
  assert.equal(yaml.includes(`  ${BASE}/pre-arrange:`), false)
})
test('grouped native protocol proves exact groups, command replay, real readback and versioned source release', async () => {
  const h = server(), proof = await verifyGroupedDispatch(h.context, 'offline', h.candidates)
  assert.equal(proof.status, 'GROUPED_DISPATCH_VERIFIED'); assert.equal(proof.cleanup, 'OWNED_UNPICKED_SOURCES_RELEASED')
  assert.ok(h.checks.every(c => c.passed)); assert.equal(h.orders.size, 2); assert.ok([...h.orders.values()].every(o => !o.items.length))
  assert.equal(h.calls.filter(c => c.path.endsWith('/drafts/pre-arrange')).length, 3)
  assert.equal(h.calls.filter(c => c.path.match(/\/items\/\d+\/remove$/)).length, 2)
  assert.ok(h.calls.every(c => c.path !== `${BASE}/pre-arrange` && !/\/pick|\/depart|\/transfer-delivery/.test(c.path)))
  assert.ok(h.calls.filter(c => c.method !== 'GET' && c.path !== `${BASE}/drafts`).every(c => Object.hasOwn(c.body, 'recordVersion')))
})
test('actual grouping receipt rejects merged group identity even when totals match', () => {
  const c = rows(), groups = c.map((r, n) => ({ draftContainerId: String(n), sourceKeys: [r.sourceKey] }))
  const receipt = { vehicles: c.map((r, n) => ({ draftContainerId: String(n), dispatch: { dispatchNo: `DS-${n}`, action: 'CREATE', status: 'ACTIVE', stage: 'BEFORE_LOADING', counts: { itemCount: 1, palletCount: 1, packageCount: r.availablePackages } } })) }
  assert.equal(groupedDispatchMatches(receipt, groups, c), true)
  receipt.vehicles[1].dispatch.dispatchNo = receipt.vehicles[0].dispatch.dispatchNo; assert.equal(groupedDispatchMatches(receipt, groups, c), false)
})
test('missing marker, source version, physical ownership or ambiguous FBA allocation is NOT_RUN before writes', async () => {
  for (const mutate of [h => { h.context.createdResources.phase4OverseasPda.marker = 'OLD' }, h => { delete h.candidates[0].sourceVersion }, h => { h.candidates[0].customerId = '999' }, h => { h.candidates[0].warehouseId = '999' }, h => { h.candidates[0].appointmentBatches = [{}, {}] }]) {
    const h = server(); mutate(h); assert.equal((await verifyGroupedDispatch(h.context, 'offline', h.candidates)).status, 'NOT_RUN'); assert.equal(h.calls.length, 0)
  }
})
test('FBA allocations preserve actual BOOKING source and legal version zero; missing version is never guessed', () => {
  const c = rows()[0]; c.appointmentBatches = [{ batchKey: 'FBA_APPOINTMENT:8', customerId: '55', recordVersion: 0, sources: [{ sourceKey: 'BOOKING:ACTUAL', destinationKey: 'FBA:A1', packageCount: 1 }] }]
  const body = createBody('CURRENT', 'FBA', [c]); assert.deepEqual(body.batchAllocations[0], { planPalletId: '21', batchKey: 'FBA_APPOINTMENT:8', customerId: '55', sourceKey: 'BOOKING:ACTUAL', destinationKey: 'FBA:A1', packageCount: 1, recordVersion: 0 })
  delete c.appointmentBatches[0].recordVersion; assert.throws(() => createBody('CURRENT', 'FBA', [c]), /权威recordVersion/)
  assert.equal(version(0), 0); for (const missing of [null, undefined, '', false, -1]) assert.throws(() => version(missing))
  assert.equal(version('9223372036854775807'), '9223372036854775807'); assert.equal(versionAfter('9007199254740993', '9007199254740992'), true)
  assert.equal(versionAfter('9223372036854775806', '9223372036854775807'), false)
  assert.throws(() => version('9223372036854775808')); assert.throws(() => version(9007199254740992))
  assert.throws(() => sourceSelection({ planPalletId: '21' }), /原生来源/)
})
test('rejection side effects and merged receipts stop without mutating or cleaning more resources', async () => {
  for (const fault of ['reject-mutates-draft', 'merged-groups']) {
    const h = server(fault); await assert.rejects(verifyGroupedDispatch(h.context, 'offline', h.candidates))
    assert.equal(h.calls.filter(c => /\/items\/\d+\/remove$/.test(c.path)).length, 0)
    assert.equal(h.context.createdResources.groupedDispatch.cleanup, 'PRESERVED_FOR_REVIEW')
  }
})
test('unknown converted write keeps original request keys and forbids fresh writes across sessions', async () => {
  const h = server('unknown-convert'); await assert.rejects(verifyGroupedDispatch(h.context, 'offline', h.candidates))
  assert.equal(h.context.createdResources.groupedDispatch.status, 'OUTCOME_UNKNOWN'); assert.equal(h.orders.size, 2)
  assert.equal(h.calls.filter(c => c.path.endsWith('/drafts/pre-arrange')).length, 2); assert.equal(h.calls.filter(c => /\/remove$/.test(c.path)).length, 0)
  const io = dispatchIo(h.context, 'offline', {}, 'ANOTHER'); await assert.rejects(io.command('POST', BASE, {}, 'forbidden'), /停写/)
})
test('checkpoint failure before send causes no HTTP; callback records original intent before a successful send', async () => {
  const context = { marker: 'C', createdResources: {}, request: async () => { throw Error('must not send') }, checkpoint: async () => { throw Error('disk full') } }
  const proof = {}, io = dispatchIo(context, 'offline', proof, 'CHECKPOINT'); await assert.rejects(io.command('POST', BASE, { idempotencyKey: 'same' }, 'create'), /disk full/)
  assert.equal(proof.requests[0].status, 'NOT_SENT')
  let sent = false
  const noPersistence = { marker: 'NO_DISK', createdResources: {}, request: async () => { sent = true } }
  await assert.rejects(dispatchIo(noPersistence, 'offline', {}, 'NO_DISK').send('POST', BASE, { idempotencyKey: 'must-not-send' }, 'blocked'), /持久化checkpoint/); assert.equal(sent, false)
  const phases = [], other = { marker: 'D', createdResources: {}, checkpoint: async () => phases.push('checkpoint'), request: async () => { phases.push('request'); return { http: 200, body: { code: '200', data: { recorded: true } } } } }
  await dispatchIo(other, 'offline', {}, 'CHECKPOINT').command('POST', BASE, { idempotencyKey: 'same' }, 'create'); assert.deepEqual(phases, ['checkpoint', 'request', 'checkpoint'])
})

test('5xx, transport failure and malformed success stop writes while preserving original diagnostic evidence', async () => {
  for (const response of [{ http: 503, raw: 'upstream failure', body: null }, { http: 0, raw: 'connection reset', transportError: { code: 'ECONNRESET' } }, { http: 200, raw: '<html>unreadable receipt</html>', body: null }, { http: 200, body: { code: '200', data: null } }]) {
    let sent = 0
    const context = { marker: 'UNKNOWN', createdResources: {}, checkpoint: async () => {}, request: async () => { sent++; return response } }, proof = {}, io = dispatchIo(context, 'offline', proof, 'UNKNOWN')
    await io.send('POST', BASE, { idempotencyKey: 'original-key' }, 'original create')
    assert.equal(proof.outcomeUnknown, true); assert.equal(proof.requests[0].status, 'UNKNOWN_RESULT'); assert.equal(proof.requests[0].body.idempotencyKey, 'original-key')
    assert.equal(proof.requests[0].raw, response.raw); assert.deepEqual(proof.requests[0].transportError, response.transportError)
    await assert.rejects(io.command('POST', BASE, { idempotencyKey: 'fresh-key' }, 'forbidden'), /停写/); assert.equal(sent, 1)
  }
})