const core = require('../../core')

const { query } = core

// Raw SQL for the Asset Bridge's three tables (docs/link/v8.md §6, §8, §12).
//
// Unlike `shard_clilocs` and the atlas tables, these are NOT import-owned in the
// empty-and-refill sense, and the difference is the whole reason phase 3 put them
// in their own tables rather than in columns on `shard_spawn_creatures`.
//
// An asset row is expensive to obtain — a decode on the shard, a PNG across the
// wire, a file written under uploads/ — and it is valid until the operator
// patches their client. An atlas refresh, by contrast, happens on every boot and
// destroys everything it owns. Putting the two in one table would mean a routine
// re-parse of the ServUO tree silently deleting every imported portrait, with the
// next Update reporting "nothing changed" and never restoring them.
//
// So these are upserted per key, and the only thing that ever deletes from them
// is an explicit removal of a key the shard no longer offers — which is staged
// for review, never applied silently (§6).

const BATCH = 500

async function batched(conn, sql, rows) {
  for (let i = 0; i < rows.length; i += BATCH) {
    await conn.batch(sql, rows.slice(i, i + BATCH))
  }
  return rows.length
}

// ── the manifest side ──────────────────────────────────────────────────────

/**
 * The asset rows we hold in one family, as a Map of key → row.
 *
 * **The family is required, and the reason is a deletion.** The import diffs what
 * this returns against a manifest, and a manifest is always of ONE family (§14 —
 * the reply carries a single catalogue id, so it could not be otherwise). Phase 5
 * put item and land art in this table beside the body catalogue; read whole, the
 * body import then sees every item picture as a key the shard has stopped
 * offering and stages all of them for deletion. On a real install that is a few
 * hundred pictures the operator is asked to approve the loss of, with a sentence
 * that is entirely wrong about what happened.
 *
 * `null` reads every family, which nothing in the import path should ever want.
 */
async function allAssets(family = null) {
  const rows = await query(
    'SELECT asset_key, family, sha256, bytes, width, height, body, action, direction, file, catalog ' +
      'FROM shard_assets' +
      (family ? ' WHERE family = ?' : ''),
    family ? [family] : [],
  )

  const map = new Map()

  for (const row of rows) {
    map.set(row.asset_key, {
      key: row.asset_key,
      family: row.family,
      sha256: row.sha256,
      bytes: Number(row.bytes) || 0,
      width: Number(row.width) || 0,
      height: Number(row.height) || 0,
      body: row.body === null ? null : Number(row.body),
      action: row.action === null ? null : Number(row.action),
      direction: row.direction === null ? null : Number(row.direction),
      file: row.file || null,
      catalog: row.catalog || null,
    })
  }

  return map
}

/**
 * Write the assets an import produced, and record what the import was.
 *
 * One transaction for the rows and the meta together: the meta row is what an
 * Update compares against to decide there is nothing to do, so a meta written
 * without its rows would make the site believe it holds a catalogue it does not.
 *
 * `ON DUPLICATE KEY UPDATE` rather than delete-and-insert, because an unchanged
 * key must keep the file it already points at — re-writing the file for every
 * asset on every Update is exactly the cost the manifest diff exists to avoid.
 *
 * `remove` is the keys an operator has APPROVED the loss of (§6). They are
 * deleted here, inside the same transaction, because a half-applied removal is
 * the worst of the three outcomes: until phase 8 the import unlinked the sprite
 * and left the row, so the catalogue still counted a picture that was gone, the
 * atlas could point a creature at a deleted file, and the very next forced
 * import staged the same key for review again — telling the operator nothing had
 * changed, about a file it had already deleted.
 */
async function saveAssets(rows, meta, remove = []) {
  const conn = await core.pool.getConnection()

  try {
    await conn.beginTransaction()

    const values = rows.map((r) => [
      r.key,
      r.family || 'body',
      r.sha256,
      r.bytes ?? 0,
      r.width ?? 0,
      r.height ?? 0,
      r.body ?? null,
      r.action ?? null,
      r.direction ?? null,
      r.file ?? null,
      r.catalog ?? meta?.catalog ?? null,
    ])

    await batched(
      conn,
      'INSERT INTO shard_assets ' +
        '(asset_key, family, sha256, bytes, width, height, body, action, direction, file, catalog) ' +
        'VALUES (?,?,?,?,?,?,?,?,?,?,?) ' +
        'ON DUPLICATE KEY UPDATE family = VALUES(family), sha256 = VALUES(sha256), ' +
        'bytes = VALUES(bytes), width = VALUES(width), height = VALUES(height), ' +
        'body = VALUES(body), action = VALUES(action), direction = VALUES(direction), ' +
        'file = VALUES(file), catalog = VALUES(catalog), imported_at = CURRENT_TIMESTAMP',
      values,
    )

    if (remove.length > 0) {
      for (let i = 0; i < remove.length; i += BATCH) {
        const slice = remove.slice(i, i + BATCH)
        await conn.query(
          `DELETE FROM shard_assets WHERE asset_key IN (${slice.map(() => '?').join(',')})`,
          slice,
        )
      }
    }

    if (meta) {
      await conn.query(
        'INSERT INTO shard_asset_meta (id, payload) VALUES (1, ?) ' +
          'ON DUPLICATE KEY UPDATE payload = VALUES(payload), imported_at = CURRENT_TIMESTAMP',
        [JSON.stringify(meta)],
      )
    }

    await conn.commit()

    return values.length
  } catch (err) {
    await conn.rollback().catch(() => {})
    throw err
  } finally {
    conn.release()
  }
}

// ── the on-demand side (§11, phase 5) ──────────────────────────────────────

/**
 * The pictures we hold for an explicit list of keys, as a Map of key → filename.
 *
 * This is the read on the hot path — every marketplace page and every character
 * sheet runs it — so it is one statement over the primary key and it returns only
 * what it is asked for. It deliberately does NOT check staleness: a page renders
 * the picture it has, and deciding whether that picture is out of date is the warm
 * pass's job, off the request.
 */
async function filesForKeys(keys) {
  const list = [...new Set(keys.filter((k) => typeof k === 'string' && k !== ''))]

  if (list.length === 0) return new Map()

  const rows = await query(
    `SELECT asset_key, file FROM shard_assets WHERE file IS NOT NULL AND asset_key IN (${list
      .map(() => '?')
      .join(',')})`,
    list,
  )

  const map = new Map()

  for (const row of rows) map.set(row.asset_key, row.file)

  return map
}

/**
 * Which of these keys we already hold under the shard's CURRENT catalogue.
 *
 * The warm pass subtracts this from what it wants, so everything it does not
 * return gets fetched: a key we have never seen, and a key whose row was written
 * against a catalogue the shard has since moved past (§7 — an operator patched
 * their client). A row with no file is not held either, because the database and
 * the uploads volume can disagree and a broken image is worse than a re-fetch.
 */
async function freshKeys(keys, catalog) {
  const list = [...new Set(keys.filter((k) => typeof k === 'string' && k !== ''))]

  if (list.length === 0) return new Set()

  const rows = await query(
    `SELECT asset_key FROM shard_assets WHERE file IS NOT NULL AND catalog <=> ? ` +
      `AND asset_key IN (${list.map(() => '?').join(',')})`,
    [catalog ?? null, ...list],
  )

  return new Set(rows.map((r) => r.asset_key))
}

/** Counts for the admin surface, split by family. */
async function countByFamily() {
  const rows = await query(
    'SELECT family, COUNT(*) AS total, SUM(file IS NOT NULL) AS stored FROM shard_assets GROUP BY family',
  )

  const out = {}

  for (const row of rows) {
    out[row.family] = { total: Number(row.total) || 0, stored: Number(row.stored) || 0 }
  }

  return out
}

/**
 * Record what the import that just finished actually did (§6, phase 8).
 *
 * **A second write, deliberately.** The interesting half of that summary — how
 * many atlas creatures resolved to a body id, how many portraits were applied —
 * does not exist when `saveAssets` commits: producing it takes another round trip
 * to the shard, and widening the rows-and-meta transaction to cover a network
 * call is how an import ends up holding a write lock for the length of a timeout.
 *
 * `JSON_SET` rather than a read-modify-write for the same reason the rest of this
 * file is one statement per operation: the payload is the gate an Update compares
 * against, and re-serialising it from the outside is how a concurrent import
 * loses a field nobody notices for a month.
 *
 * It is cosmetic by design — nothing reads `last` to make a decision, the panel
 * only renders it — so a failure here is logged and swallowed by the caller
 * rather than failing an import that has already applied.
 */
async function recordLastImport(last) {
  await query('UPDATE shard_asset_meta SET payload = JSON_SET(payload, ?, JSON_COMPACT(?)) WHERE id = 1', [
    '$.last',
    JSON.stringify(last),
  ])
}

async function getMeta() {
  const rows = await query('SELECT payload, imported_at FROM shard_asset_meta WHERE id = 1')
  if (rows.length === 0) return null
  const payload = typeof rows[0].payload === 'string' ? JSON.parse(rows[0].payload) : rows[0].payload
  return { ...payload, importedAt: rows[0].imported_at }
}

/**
 * How many assets we hold, optionally in one family.
 *
 * **The family argument is not optional in spirit.** Phase 5 put item and land
 * art in this table beside the body catalogue, and they are counted differently
 * by nature: the catalogue is a SET with a known size, while item art is however
 * much of an unbounded space the site has happened to ask for. A whole-table
 * count answers neither question — it reported the creature catalogue as 1,408
 * rows on an install holding 1,095 portraits and 313 item pictures, which is a
 * confident wrong number in the one place an operator checks whether the import
 * worked.
 */
async function countAssets(family = null) {
  const rows = await query(
    'SELECT COUNT(*) AS n, SUM(file IS NOT NULL) AS stored FROM shard_assets' +
      (family ? ' WHERE family = ?' : ''),
    family ? [family] : [],
  )
  return { total: Number(rows[0]?.n) || 0, stored: Number(rows[0]?.stored) || 0 }
}

// ── the body resolution side (§8) ──────────────────────────────────────────

/**
 * Replace the whole slug → body map.
 *
 * This one IS a replace, and for the opposite reason to the assets above: it is
 * derived from the atlas's creature list, so a slug that has left the atlas has
 * no meaning any more and keeping its row would leave the map growing forever
 * across map changes. The pass that produces it is cheap to redo — a shard round
 * trip, no files — which is what makes replacing safe here and not there.
 */
async function replaceBodies(rows) {
  const conn = await core.pool.getConnection()

  try {
    await conn.beginTransaction()
    await conn.query('DELETE FROM shard_creature_bodies')

    const values = rows.map((r) => [r.slug, r.typeName, r.body ?? null, r.status || 'ok'])

    await batched(
      conn,
      'INSERT INTO shard_creature_bodies (slug, type_name, body, status) VALUES (?,?,?,?)',
      values,
    )

    await conn.commit()

    return values.length
  } catch (err) {
    await conn.rollback().catch(() => {})
    throw err
  } finally {
    conn.release()
  }
}

async function allBodies() {
  return query(
    'SELECT slug, type_name, body, status, resolved_at FROM shard_creature_bodies ORDER BY slug',
  )
}

async function countBodies() {
  const rows = await query(
    "SELECT COUNT(*) AS n, SUM(status = 'ok') AS resolved FROM shard_creature_bodies",
  )
  return { total: Number(rows[0]?.n) || 0, resolved: Number(rows[0]?.resolved) || 0 }
}

/**
 * The derivation `replaceAtlas` applies on the way past: slug → uploaded filename.
 *
 * One join rather than two reads, because it runs inside the atlas transaction —
 * the atlas rows are being inserted at that moment and every extra round trip is
 * time the site's creature list does not exist.
 *
 * Rows with no body, no asset or an asset whose bytes were never fetched are
 * simply absent from the result, which is what leaves `art` NULL. That is a
 * first-class state everywhere it is consumed and the expected one for two thirds
 * of the player bodies (§5.2).
 *
 * **The join is pinned to the catalogue key, not merely to the body id** — and as
 * of phase 6 that key is no longer always `a0`. 73 of this client's bodies have
 * no art at action 0 and are catalogued at the first action that does (§11.2), so
 * a join hardcoding `a0` would silently drop exactly the creatures this phase
 * added — a horse among them. It reads the row's own `action` instead, which
 * still excludes any deeper key a later phase adds (`body/400/a2/f0` does not
 * equal `body/400/a2`), so one slug still matches at most one row.
 *
 * `COALESCE(a.action, 0)` because a row written before this column existed has
 * NULL there and a NULL inside `CONCAT` makes the whole comparison NULL — which
 * would have dropped every portrait on the site until the next import, with the
 * database perfectly correct.
 */
async function artBySlug() {
  const rows = await query(
    'SELECT b.slug, a.file FROM shard_creature_bodies b ' +
      "JOIN shard_assets a ON a.body = b.body AND a.family = 'body' " +
      "AND a.asset_key = CONCAT('body/', b.body, '/a', COALESCE(a.action, 0)) " +
      "WHERE b.status = 'ok' AND b.body IS NOT NULL AND a.file IS NOT NULL",
  )

  const map = {}

  for (const row of rows) map[row.slug] = row.file

  return map
}

module.exports = {
  allAssets,
  saveAssets,
  recordLastImport,
  getMeta,
  countAssets,
  replaceBodies,
  allBodies,
  countBodies,
  artBySlug,
  filesForKeys,
  freshKeys,
  countByFamily,
}