// The Asset Bridge client (docs/link/v8.md §5, §6, §8 — protocol 8, phase 3).
//
// Three walks over the same request/reply path `clilocBridge.js` already uses,
// and everything that file says about the envelope holds here unchanged: only
// `cut: 'end'` means finished, the cursor must advance, and 425 is the ordinary
// answer during an import rather than an error.
//
// What is different is what each walk is FOR.
//
// ── `readManifest` — what the shard could serve, without the pixels ────────
//
// §6's stage 2. Every row is `{ key, sha256, bytes, width, height }`, so the
// site can diff against what it already holds and ask for only the keys whose
// hash moved. On the ordinary case — a shard restart that changed nothing —
// that diff is empty and no pixels cross at all.
//
// This family pages on the shard's WALL CLOCK, not on bytes. Its rows are about
// ninety bytes and the whole catalogue is one page by the byte budget, but
// producing that page means decoding hundreds of sprites and the sidecar waits
// ten seconds for a reply. So `cut: 'limit'` is the normal page ending here,
// where for clilocs it would have signalled something wrong.
//
// ── `fetchAssets` — the pixels, for keys we chose ─────────────────────────
//
// Each row carries a base64 PNG. The shard encodes it: `System.Drawing` is
// already in its decode path, so PNG costs it no new dependency, and having the
// hash cover exactly the bytes we store is what makes the next Update a diff.
//
// **`catalog` is passed on every fetch and it is not optional in practice.** It
// is an id the shard derives from the client files themselves, so handing it back
// makes the shard refuse if those files moved since the manifest was read.
// Without it an operator patching their client mid-import produces one asset set
// stitched out of two, with no error anywhere — the same failure `clilocBridge`
// guards against by comparing (size, mtime) across pages.
//
// ── `resolveBodies` — the atlas's creatures, by class name ────────────────
//
// §8. The shard constructs each type and reads `Body.BodyID`, which is the only
// thing that is correct for a shard's own custom creatures. That runs on its Core
// thread, so the batch is small and the shard REFUSES an over-long list rather
// than truncating it — hence the chunking here, and hence a chunk size that is a
// constant rather than "as many as fit".

// 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 uoLinkClient = require('./uoLinkClient')
const log = require('../core').logger('asset-bridge')

/** The only family phase 3 serves. §5's key scheme covers statics and land later. */
const FAMILY = 'body'

// Chunk size for the body pass. The shard's own cap defaults to 100 and it
// refuses rather than truncates, so this must stay at or under it — a mismatch
// here does not degrade, it fails every chunk.
const BODY_CHUNK = 100

// Chunk size for a fetch request. The shard cuts the PAGE by byte budget within
// whatever it is handed, so this only bounds how large a single request is; a
// chunk of 400 one-kilobyte sprites is a couple of pages.
const FETCH_CHUNK = 400

// Bounds on each walk. None is expected to be reached — the catalogue is under a
// thousand rows — and each exists so that a shard answering nonsense costs a
// bounded amount of time rather than an unbounded amount of memory.
const MAX_PAGES = 200
const MAX_ROWS = 100000

// 425 is flow control, not failure: the shard's asset plane serves one request at
// a time because its outbound queue is bounded in lines rather than bytes. During
// an import a page coming back busy is expected, so it is retried with a backoff
// rather than failing the walk.
const BUSY_RETRIES = 6
const BUSY_BACKOFF_MS = [200, 400, 800, 1600, 3200, 5000]

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

const sleep = (ms) => new Promise((resolve) => setTimeout(resolve, ms))

/**
 * Map a sidecar response onto one of this module's codes.
 *
 * Deliberately the same vocabulary `clilocBridge.describeFailure` uses, because
 * the admin panel reports them side by side and an operator should not have to
 * learn two names for "you have not switched this on".
 *
 * 422 is the one that means something different here: on the cliloc path it is a
 * file the shard cannot decode, and on this one it is *also* the mid-import guard
 * firing — the client files moved between the manifest and the fetch.
 */
function describeFailure(res, what) {
  const reason = res?.data?.reason || res?.error || `sidecar responded ${res?.status}`

  switch (res?.status) {
    case 403:
      return new AssetBridgeError(
        `The shard is refusing to serve client assets (Bridge.AssetsEnabled is off): ${reason}`,
        'DISABLED',
      )
    case 404:
      return new AssetBridgeError(`The shard has no ${what}: ${reason}`, 'NO_SOURCE')
    case 409:
      return new AssetBridgeError(
        `The sidecar refused the protocol version this build declares: ${reason}`,
        'PROTOCOL',
      )
    case 422:
      return new AssetBridgeError(reason, 'SOURCE_CHANGED')
    case 425:
      return new AssetBridgeError(
        'The shard stayed busy serving another asset request',
        'BUSY',
      )
    case 503:
      // The named `NO_IMAGING` outcome arrives this way: a Linux shard host with
      // no libgdiplus cannot render a sprite at all, and §4.4 requires that be an
      // actionable sentence rather than a stack trace. The shard's own wording
      // already names the package and the command, so it is passed through.
      return new AssetBridgeError(reason, /libgdiplus/i.test(reason) ? 'NO_IMAGING' : 'SHARD_DOWN')
    case 504:
      return new AssetBridgeError(`The shard did not answer: ${reason}`, 'SHARD_DOWN')
    default:
      return new AssetBridgeError(reason, 'UNAVAILABLE')
  }
}

/** One call, with the 425 backoff. `send` returns the client's `{ ok, ... }`. */
async function withBusyRetry(send, what) {
  for (let attempt = 0; ; attempt++) {
    const res = await send()
    if (res.ok) return res.data

    if (res.status === 425 && attempt < BUSY_RETRIES) {
      await sleep(BUSY_BACKOFF_MS[Math.min(attempt, BUSY_BACKOFF_MS.length - 1)])
      continue
    }

    throw describeFailure(res, what)
  }
}

/**
 * Shared page-envelope checks (§3.4).
 *
 * Every one of these is a way a walk can end in something that LOOKS like a
 * complete import and is not, which is why they are assertions rather than
 * warnings: a truncated catalogue is indistinguishable downstream from a client
 * that simply has fewer creatures.
 */
function checkPage(page, { arrayName, cursor, pages }) {
  if (!page || !Array.isArray(page[arrayName])) {
    throw new AssetBridgeError(
      `The shard sent an asset page with no ${arrayName} array`,
      'MALFORMED',
    )
  }

  if (!page.more) {
    if (page.cut !== 'end') {
      throw new AssetBridgeError(
        `The shard stopped sending assets after ${pages} page(s) (cut: ${page.cut || 'unknown'})`,
        'INCOMPLETE',
      )
    }
    return { done: true }
  }

  if (!page.cursor || page.cursor === cursor) {
    throw new AssetBridgeError(
      `The shard asked for another asset page without advancing its cursor (${page.cursor || 'none'})`,
      'STUCK',
    )
  }

  return { done: false, cursor: page.cursor }
}

// The client files the body catalogue is derived from. `assets.sources` reports
// every file the shard can see; these are the ones that decide a sprite.
//
// `body.def` and `bodyconv.def` are in the list and it would be easy to leave
// them out — they hold no pixels. They decide WHICH record a body id resolves to,
// so an operator editing one changes what every affected creature looks like
// while every anim file stays byte-identical. That is precisely the drift a
// content hash of the art files cannot see.
const SOURCE_FILES = [
  'anim.idx', 'anim.mul',
  'anim2.idx', 'anim2.mul',
  'anim3.idx', 'anim3.mul',
  'anim4.idx', 'anim4.mul',
  'anim5.idx', 'anim5.mul',
  'body.def', 'bodyconv.def',
  'verdata.mul',
]

/**
 * Stage 1 of the import gate (§6): have the client files this family reads
 * changed at all?
 *
 * Returns `{ files, extractorVersion, hashing, complete, imaging }` where `files`
 * is a `{ name: { size, mtime, sha256 } }` map over `SOURCE_FILES` — a file the
 * shard does not have is simply absent, which is normal (few clients carry all
 * five anim files).
 *
 * **A null `sha256` means "not computed yet", never "changed".** The shard hashes
 * off the request path because `anim.mul` alone is 195 MB and hashing it cannot
 * fit inside a reply, and it reports `hashing: true` while that runs.
 * `sameSources` below falls back to (size, mtime) in that case, which is the same
 * gate the shard itself applies.
 */
async function sourceFingerprint() {
  const res = await uoLinkClient.getAssetSources()
  if (!res.ok) throw describeFailure(res, 'client file manifest')

  const wanted = new Set(SOURCE_FILES)
  const files = {}

  for (const entry of res.data?.files ?? []) {
    const name = String(entry?.name || '').toLowerCase()
    if (!wanted.has(name)) continue

    files[name] = {
      size: Number(entry.size) || 0,
      mtime: Number(entry.mtime) || 0,
      sha256: entry.sha256 ?? null,
    }
  }

  return {
    files,
    extractorVersion: Number(res.data?.extractorVersion) || 0,
    hashing: Boolean(res.data?.hashing),
    complete: Boolean(res.data?.complete),
    imaging: res.data?.imaging ?? null,
    // Which §5 key families this overlay can be asked for (phase 5). Absent on a
    // phase-3 or phase-4 overlay, which served bodies and nothing else — so the
    // fallback is `['body']` rather than `[]`: an older shard is not a shard with
    // no assets, and treating it as one would turn a working bestiary off.
    families: Array.isArray(res.data?.families) && res.data.families.length > 0
      ? res.data.families.map(String)
      : [FAMILY],
  }
}

/**
 * True when two source fingerprints describe the same client files.
 *
 * The file SET has to match as well as each file's contents: a client that gained
 * an `anim5.mul` it did not have before is a client whose gargoyles suddenly
 * resolve, and comparing only the files present in both would call that
 * unchanged.
 */
function sameSources(a, b) {
  if (!a || !b) return false
  if (a.extractorVersion !== b.extractorVersion) return false

  const names = new Set([...Object.keys(a.files ?? {}), ...Object.keys(b.files ?? {})])

  for (const name of names) {
    const left = a.files?.[name]
    const right = b.files?.[name]

    if (!left || !right) return false

    if (left.sha256 && right.sha256) {
      if (left.sha256 !== right.sha256) return false
      continue
    }

    if (left.size !== right.size || left.mtime !== right.mtime || left.size <= 0) return false
  }

  return names.size > 0
}

/**
 * Stage 2: the whole manifest for the body family.
 *
 * Returns `{ rows, catalog, extractorVersion, playerBodies, pages, scanned }`.
 * No pixels — `rows` is `[{ key, sha256, bytes, width, height, body, direction }]`.
 */
async function readManifest({ family = FAMILY } = {}) {
  const started = Date.now()
  const rows = []

  let cursor = null
  let pages = 0
  let catalog = null
  let extractorVersion = 0
  let playerBodies = []
  let scanned = 0
  let finished = false

  while (pages < MAX_PAGES) {
    const page = await withBusyRetry(
      () => uoLinkClient.getAssetManifest({ family, cursor }),
      `${family} asset manifest`,
    )
    pages++

    if (catalog === null) {
      catalog = page.catalog ?? null
      extractorVersion = Number(page.extractorVersion) || 0
      playerBodies = Array.isArray(page.playerBodies) ? page.playerBodies.map(Number) : []
    } else if (page.catalog !== catalog) {
      // The client files moved between two pages of one walk. Refusing is the
      // only honest answer: half of what we hold describes files that no longer
      // exist, and nothing later can tell which half.
      throw new AssetBridgeError(
        "The shard's client files changed while the manifest was being read; nothing was imported",
        'SOURCE_CHANGED',
      )
    }

    scanned += Number(page.scanned) || 0

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

      rows.push({
        key,
        family,
        sha256: String(row?.sha256 ?? ''),
        bytes: Number(row?.bytes) || 0,
        width: Number(row?.width) || 0,
        height: Number(row?.height) || 0,
        body: Number.isFinite(Number(row?.body)) ? Number(row.body) : null,
        // Which action the thumbnail came from (§11.2, phase 6). All but 73 of
        // this client's bodies answer 0; the rest have no art there and are
        // catalogued deeper, with the key naming the action. An overlay older
        // than phase 6 omits it, and 0 is the right reading of that.
        action: Number.isFinite(Number(row?.action)) ? Number(row.action) : 0,
        direction: Number.isFinite(Number(row?.direction)) ? Number(row.direction) : null,
      })
    }

    if (rows.length > MAX_ROWS) {
      throw new AssetBridgeError(
        `The shard listed more than ${MAX_ROWS} assets; refusing to keep reading`,
        'TOO_LARGE',
      )
    }

    const state = checkPage(page, { arrayName: 'rows', cursor, pages })

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

    cursor = state.cursor
  }

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

  log.info('asset manifest read from the shard', {
    family,
    rows: rows.length,
    scanned,
    pages,
    ms: Date.now() - started,
  })

  return { rows, catalog, extractorVersion, playerBodies, pages, scanned }
}

/**
 * The bytes for an explicit list of keys.
 *
 * Returns a Map of key → `{ sha256, bytes, width, height, body, action, direction, png }`
 * where `png` is a Buffer. A key the shard could not serve is **absent from the
 * map** rather than present with a null — the caller then decides what that means
 * for its own row, and the two ways it happens (`absent`, `unsupported`) are
 * counted separately in the returned tallies so an operator can tell "this client
 * has no art for that body" from "the site asked for a key shape this shard does
 * not serve", which is a bug rather than a gap.
 */
async function fetchAssets({ keys, catalog } = {}) {
  const started = Date.now()
  const out = new Map()
  const missing = { absent: 0, unsupported: 0 }

  const list = Array.isArray(keys) ? keys.filter((k) => typeof k === 'string' && k !== '') : []

  if (list.length === 0) return { assets: out, missing, pages: 0, catalog: catalog ?? null }

  let pages = 0

  // The catalogue the shard actually answered under. The body import already knows
  // it from the manifest, but the on-demand families have no manifest to learn it
  // from (§11) — so it is read back off the reply and stored with the rows, which
  // is what makes a later "is this stale?" answerable per key.
  let answered = catalog ?? null

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

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

    while (walked < MAX_PAGES) {
      const page = await withBusyRetry(
        () => uoLinkClient.fetchAssets({ keys: chunk, catalog, cursor }),
        'asset content',
      )
      pages++
      walked++

      if (typeof page.catalog === 'string' && page.catalog !== '') {
        if (answered !== null && page.catalog !== answered) {
          // Two pages of one walk describing two different clients. The shard
          // refuses this when it is told what to expect; when it was not told —
          // the first fetch of a warm pass — this is where it is caught.
          throw new AssetBridgeError(
            `The shard's client files changed mid-fetch (catalog ${answered} became ${page.catalog})`,
            'UNAVAILABLE',
          )
        }

        answered = page.catalog
      }

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

        if (row?.status !== 'ok') {
          if (row?.status === 'unsupported') missing.unsupported++
          else missing.absent++
          continue
        }

        if (typeof row.png !== 'string' || row.png === '') {
          missing.absent++
          continue
        }

        out.set(key, {
          sha256: String(row.sha256 ?? ''),
          bytes: Number(row.bytes) || 0,
          width: Number(row.width) || 0,
          height: Number(row.height) || 0,
          body: Number.isFinite(Number(row.body)) ? Number(row.body) : null,
          action: Number.isFinite(Number(row.action)) ? Number(row.action) : null,
          direction: Number.isFinite(Number(row.direction)) ? Number(row.direction) : null,
          // Phase 5's art families carry these; the body catalogue does not, and a
          // consumer that wants neither is unaffected by either.
          hue: Number.isFinite(Number(row.hue)) ? Number(row.hue) : null,
          partialHue: typeof row.partialHue === 'boolean' ? row.partialHue : null,
          source: typeof row.source === 'string' ? row.source : null,
          png: Buffer.from(row.png, 'base64'),
        })
      }

      const state = checkPage(page, { arrayName: 'rows', cursor, pages: walked })

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

      cursor = state.cursor
    }

    if (!finished) {
      throw new AssetBridgeError(
        `An asset fetch did not end within ${MAX_PAGES} pages; nothing was imported`,
        'TOO_LARGE',
      )
    }
  }

  log.info('asset content fetched from the shard', {
    catalog: answered,
    asked: list.length,
    got: out.size,
    absent: missing.absent,
    unsupported: missing.unsupported,
    pages,
    ms: Date.now() - started,
  })

  return { assets: out, missing, pages, catalog: answered }
}

/**
 * Slug → body id, for the atlas's own creature list (§8).
 *
 * `creatures` is `[{ slug, name }]` where `name` is the ServUO class name — which
 * `shard_spawn_creatures.name` already holds, because the atlas build picks the
 * winning spelling of the spawn TYPE token rather than inventing a display name.
 * That is why this needs no new column to ask its question.
 *
 * Returns `[{ slug, typeName, body, status }]`, one row per creature asked, with
 * every outcome recorded — including the negative ones. A creature the shard says
 * it does not have is a fact worth keeping: without it, the next pass asks again,
 * and the pass costs a real constructor per name on the shard's Core thread.
 */
async function resolveBodies({ creatures } = {}) {
  const started = Date.now()
  const list = Array.isArray(creatures) ? creatures : []
  const out = []

  for (let i = 0; i < list.length; i += BODY_CHUNK) {
    const chunk = list.slice(i, i + BODY_CHUNK)
    const bySlug = new Map()

    for (const creature of chunk) {
      const typeName = String(creature?.name ?? '').trim()
      if (typeName === '') continue
      // Several slugs can share a type name only if the atlas slugified two
      // spellings to one slug, in which case they ARE one creature; asking once
      // per distinct name is what keeps the batch inside the shard's cap.
      if (!bySlug.has(typeName)) bySlug.set(typeName, [])
      bySlug.get(typeName).push(String(creature.slug))
    }

    const types = [...bySlug.keys()]
    if (types.length === 0) continue

    const page = await withBusyRetry(() => uoLinkClient.resolveBodies(types), 'body resolution')

    if (!page || !Array.isArray(page.rows)) {
      throw new AssetBridgeError('The shard sent a body resolution with no rows array', 'MALFORMED')
    }

    for (const row of page.rows) {
      const typeName = String(row?.type ?? '')
      const slugs = bySlug.get(typeName)

      if (!slugs) continue

      const status = String(row?.status ?? 'failed')
      const body = status === 'ok' && Number.isFinite(Number(row?.body)) ? Number(row.body) : null

      for (const slug of slugs) out.push({ slug, typeName, body, status })
    }
  }

  const resolved = out.filter((r) => r.status === 'ok').length

  log.info('creature bodies resolved by the shard', {
    asked: list.length,
    answered: out.length,
    resolved,
    ms: Date.now() - started,
  })

  return out
}

module.exports = {
  AssetBridgeError,
  FAMILY,
  BODY_CHUNK,
  FETCH_CHUNK,
  MAX_PAGES,
  MAX_ROWS,
  SOURCE_FILES,
  sourceFingerprint,
  sameSources,
  readManifest,
  fetchAssets,
  resolveBodies,
}