// Spawn atlas sources — the SHARD half (docs/link/v8.md §10, protocol 8 phase 7).
//
// `spawnAtlasSource.js` reads a ServUO tree off a filesystem and has done since
// the atlas existed. That is the half this replaces, and it is worth being blunt
// about what was wrong with it: `SPAWN_ATLAS.md` required the WEBSITE to be able
// to read the shard's directories — "the same host, a bind mount, or a shared
// volume". Everything else about this platform holds that only the sidecar
// bridges the shard, and that one requirement broke the rule using the component
// that faces the internet.
//
// So the shard now serves its own files over the same request/reply path as
// every other shard read, and `SERVUO_PATH` becomes what it should always have
// been: the development and same-host convenience, not the design.
//
// **The parsers do not move.** `spawnAtlasParse.js` is pure, fs-free and covered
// by CI without a ServUO tree anywhere near it, and every quirk it handles — the
// two respawn delay units, `:OBJ=` splitting, facet-name reconciliation, the
// XmlSpawner directive stripping — stays exactly where it is. The shard sends
// bytes; the website still decides what they mean.
//
// ── Why a file arrives in pieces ──────────────────────────────────────────
//
// §10 said the shard would serve `tree/<label>` → bytes, and phase 7 measured
// that it cannot. A stock `Spawns/trammel.xml` is 4.03 MB; the sidecar discards
// any inbound line over 1 MiB; that file as a single base64 row is 5.4 MiB. It
// would never arrive — the reply would be dropped, the request would time out,
// and the import would retry forever with no error in it anywhere. Two files on
// a STOCK tree are in that state.
//
// A file therefore crosses as chunks, each gzipped:
//
//   tree/Spawns/trammel.xml        the manifest row — size, hash, chunk count
//   tree/Spawns/trammel.xml/c0     the first 512 KiB of it, gzipped
//
// Measured on the stock 57.4 tree: 141 files, 11.34 MB, 158 chunks, three pages,
// 1.33 MB actually on the wire.
//
// ── The three checks below that are not decoration ────────────────────────
//
// Every one of them catches a way this can end in a tree that LOOKS imported:
//
//   - **Each chunk re-declares its own address** and carries the hash of its own
//     uncompressed bytes. A reassembly that put chunk 3 where chunk 4 belongs
//     would produce XML that still parses — XML is forgiving about what it skips
//     — and an atlas quietly missing spawns.
//   - **The whole file is hashed after reassembly** against what the manifest
//     said, which is also the fingerprint the drift gate stores.
//   - **The catalog must not move mid-walk.** An operator editing a spawn file
//     while this runs would otherwise produce one atlas stitched out of two
//     trees, with nothing anywhere reporting a problem.

// Required as a namespace, not destructured: a test that stubs the sidecar
// replaces these on the module object, and a destructured copy taken at load
// time would keep calling the real one.
const zlib = require('zlib')
const uoLinkClient = require('./uoLinkClient')
const assetBridge = require('./assetBridge')
const log = require('../core').logger('tree-bridge')

/** The §5 key family the shard serves these under. */
const FAMILY = 'tree'

// How many chunk keys go in one `assets.fetch`. The shard cuts the PAGE by byte
// budget within whatever it is asked for, so this only bounds the request; a
// stock tree's 158 chunks fit in a single one.
const FETCH_CHUNK = 200

// Bounds on the walk. Neither is expected to be reached on any real tree — the
// stock one is 141 files and 158 chunks — and both exist so that a shard
// answering nonsense costs a bounded amount of memory rather than all of it.
const MAX_FILES = 20000
const MAX_BYTES = 256 * 1024 * 1024

class TreeBridgeError extends Error {
  constructor(message, code) {
    super(message)
    this.name = 'TreeBridgeError'
    this.code = code
  }
}

/**
 * Recast an asset-plane failure as one of ours.
 *
 * The distinction worth keeping is 403: on this family it does NOT mean the
 * operator declined to serve their client files, it means they declined to serve
 * their own configuration tree — a different switch (`Bridge.TreeEnabled`) with
 * a different fix, and telling them to look at the wrong one costs them an
 * afternoon.
 */
function rethrow(err, what) {
  if (!(err instanceof assetBridge.AssetBridgeError)) return err

  if (err.code === 'DISABLED') {
    return new TreeBridgeError(
      'The shard is refusing to serve its configuration tree (Bridge.TreeEnabled is off): '
        + err.message,
      'DISABLED',
    )
  }

  return new TreeBridgeError(`${what}: ${err.message}`, err.code)
}

/**
 * Stage 1: every atlas source file the shard has, with its hash — and no bytes.
 *
 * Returns `{ catalog, chunkBytes, files: [{ key, label, bytes, mtime, chunks,
 * sha256 }] }`.
 *
 * This is the whole of the drift gate. The website stores these hashes; the next
 * import asks for this list again and fetches nothing at all when nothing moved,
 * which on a shard whose maps are not being edited is every import. One page and
 * about 32 KB on a stock tree.
 */
async function manifest() {
  const started = Date.now()
  const files = []

  let cursor = null
  let pages = 0
  let catalog = null
  let chunkBytes = 0
  let finished = false

  while (pages < assetBridge.MAX_PAGES) {
    let page

    try {
      page = await assetBridge.withBusyRetry(
        () => uoLinkClient.getAssetManifest({ family: FAMILY, cursor }),
        'the shard configuration tree',
      )
    } catch (err) {
      throw rethrow(err, 'reading the tree manifest')
    }

    pages++

    if (catalog === null) {
      catalog = page.catalog ?? null
      chunkBytes = Number(page.chunkBytes) || 0
    } else if (page.catalog !== catalog) {
      throw new TreeBridgeError(
        "The shard's configuration tree changed while it was being listed; nothing was imported",
        'SOURCE_CHANGED',
      )
    }

    for (const row of page.rows ?? []) {
      const label = String(row?.label ?? '')
      if (label === '') continue

      files.push({
        key: String(row?.key ?? `${FAMILY}/${label}`),
        label,
        bytes: Number(row?.bytes) || 0,
        mtime: Number(row?.mtime) || 0,
        chunks: Number(row?.chunks) || 0,
        sha256: row?.sha256 ? String(row.sha256) : null,
      })
    }

    if (files.length > MAX_FILES) {
      throw new TreeBridgeError(
        `The shard listed more than ${MAX_FILES} tree files; refusing to keep reading`,
        'TOO_LARGE',
      )
    }

    let state

    try {
      state = assetBridge.checkPage(page, { arrayName: 'rows', cursor, pages, noun: 'tree' })
    } catch (err) {
      throw rethrow(err, 'reading the tree manifest')
    }

    if (state.done) {
      finished = true
      break
    }

    cursor = state.cursor
  }

  if (!finished) {
    throw new TreeBridgeError(
      `The tree manifest did not end within ${assetBridge.MAX_PAGES} pages; nothing was imported`,
      'TOO_LARGE',
    )
  }

  log.info('tree manifest read from the shard', {
    files: files.length,
    catalog,
    pages,
    ms: Date.now() - started,
  })

  return { catalog, chunkBytes, files, pages }
}

/** A manifest as the `{ label: sha256 }` fingerprint the atlas model stores. */
function fingerprintOf(list) {
  const hashes = {}
  for (const file of list) hashes[file.label] = file.sha256
  return hashes
}

/**
 * Stage 2: the bytes.
 *
 * Returns `{ files: [{ label, text, sha256, bytes }] }` — deliberately the exact
 * shape `spawnAtlasSource.readSources` returns from a filesystem, so that
 * `buildAtlas` cannot tell which end a tree arrived from and nothing downstream
 * has a second code path to be wrong in.
 */
async function readSources() {
  const started = Date.now()
  const listing = await manifest()

  if (listing.files.length === 0) {
    throw new TreeBridgeError(
      'The shard served no atlas source files at all (is this a ServUO tree?)',
      'NO_SOURCE',
    )
  }

  const expected = listing.files.reduce((sum, file) => sum + file.bytes, 0)

  if (expected > MAX_BYTES) {
    throw new TreeBridgeError(
      `The shard's tree is ${expected} bytes, over the ${MAX_BYTES} this will read`,
      'TOO_LARGE',
    )
  }

  const keys = []
  for (const file of listing.files) {
    // A zero-length file is still ONE chunk. An overlay that said zero would
    // leave a manifest row nothing could ever fetch, and the walk below would
    // report the import incomplete forever.
    const chunks = Math.max(1, file.chunks)
    for (let i = 0; i < chunks; i++) keys.push(`${file.key}/c${i}`)
  }

  const parts = new Map()
  let pages = 0
  let wire = 0

  for (let i = 0; i < keys.length; i += FETCH_CHUNK) {
    const batch = keys.slice(i, i + FETCH_CHUNK)

    let cursor = null
    let finished = false
    let walked = 0

    while (walked < assetBridge.MAX_PAGES) {
      let page

      try {
        page = await assetBridge.withBusyRetry(
          () => uoLinkClient.fetchAssets({ keys: batch, catalog: listing.catalog, cursor }),
          'the shard configuration tree',
        )
      } catch (err) {
        throw rethrow(err, 'reading the tree')
      }

      pages++
      walked++

      if (typeof page.catalog === 'string' && page.catalog !== '' && page.catalog !== listing.catalog) {
        throw new TreeBridgeError(
          `The shard's configuration tree changed mid-read (catalog ${listing.catalog} became ${page.catalog})`,
          'SOURCE_CHANGED',
        )
      }

      for (const row of page.rows ?? []) {
        const key = String(row?.key ?? '')

        if (row?.status !== 'ok') {
          // Unlike an asset key, a tree key comes straight off a manifest this
          // same walk just read. There is no such thing as an expected gap here:
          // the shard listed the file, so a refusal means the tree moved or the
          // two ends disagree about the key scheme, and importing the rest would
          // silently drop whatever that file held.
          throw new TreeBridgeError(
            `The shard refused ${key || 'a tree chunk'} (${row?.status || 'unknown'}: `
              + `${row?.reason || 'no reason given'})`,
            'INCOMPLETE',
          )
        }

        const label = String(row.label ?? '')
        const chunk = Number(row.chunk)

        if (label === '' || !Number.isInteger(chunk) || chunk < 0) {
          throw new TreeBridgeError(`The shard sent a tree chunk with no address (${key})`, 'MALFORMED')
        }

        let raw

        try {
          raw = zlib.gunzipSync(Buffer.from(String(row.gzip ?? ''), 'base64'))
        } catch (err) {
          throw new TreeBridgeError(`Could not decompress ${key}: ${err.message}`, 'MALFORMED')
        }

        const declared = Number(row.bytes)

        if (Number.isFinite(declared) && declared !== raw.length) {
          throw new TreeBridgeError(
            `${key} declared ${declared} bytes and decompressed to ${raw.length}`,
            'MALFORMED',
          )
        }

        if (row.sha256 && assetBridge.sha256Of(raw) !== String(row.sha256)) {
          throw new TreeBridgeError(`${key} does not match its own hash`, 'MALFORMED')
        }

        if (!parts.has(label)) parts.set(label, new Map())
        parts.get(label).set(chunk, raw)
      }

      let state

      try {
        state = assetBridge.checkPage(page, { arrayName: 'rows', cursor, pages: walked, noun: 'tree' })
      } catch (err) {
        throw rethrow(err, 'reading the tree')
      }

      if (state.done) {
        finished = true
        break
      }

      cursor = state.cursor
    }

    if (!finished) {
      throw new TreeBridgeError(
        `A tree fetch did not end within ${assetBridge.MAX_PAGES} pages; nothing was imported`,
        'TOO_LARGE',
      )
    }
  }

  const files = []

  for (const file of listing.files) {
    const chunks = Math.max(1, file.chunks)
    const held = parts.get(file.label)

    if (!held) {
      throw new TreeBridgeError(`The shard sent nothing for ${file.label}`, 'INCOMPLETE')
    }

    const ordered = []

    for (let i = 0; i < chunks; i++) {
      const part = held.get(i)

      // Indexed rather than appended in arrival order. The rows come back in the
      // order they were asked for today, and a design that depends on that is one
      // reordering away from an atlas that is wrong in a way nothing reports.
      if (!part) {
        throw new TreeBridgeError(`${file.label} is missing chunk ${i} of ${chunks}`, 'INCOMPLETE')
      }

      ordered.push(part)
    }

    const whole = Buffer.concat(ordered)
    const sha256 = assetBridge.sha256Of(whole)

    if (file.sha256 && sha256 !== file.sha256) {
      throw new TreeBridgeError(
        `${file.label} does not match the hash its manifest row carried`,
        'MALFORMED',
      )
    }

    files.push({
      label: file.label,
      text: whole.toString('utf8'),
      sha256,
      bytes: whole.length,
    })

    wire += whole.length
  }

  log.info('tree read from the shard', {
    files: files.length,
    bytes: wire,
    chunks: keys.length,
    pages,
    ms: Date.now() - started,
  })

  return { files, catalog: listing.catalog, pages, chunks: keys.length }
}

module.exports = {
  TreeBridgeError,
  FAMILY,
  FETCH_CHUNK,
  MAX_FILES,
  MAX_BYTES,
  manifest,
  fingerprintOf,
  readSources,
}