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 ──────────────────────────────────────────────────────

/** Every asset row we hold, as a Map of key → row. */
async function allAssets() {
  const rows = await query(
    'SELECT asset_key, family, sha256, bytes, width, height, body, direction, file, catalog FROM shard_assets',
  )

  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),
      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.
 */
async function saveAssets(rows, meta) {
  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.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, 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), direction = VALUES(direction), file = VALUES(file), ' +
        'catalog = VALUES(catalog), imported_at = CURRENT_TIMESTAMP',
      values,
    )

    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
}

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 }
}

async function countAssets() {
  const rows = await query(
    'SELECT COUNT(*) AS n, SUM(file IS NOT NULL) AS stored FROM shard_assets',
  )
  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.** Today
 * one body has exactly one asset, so `a.body = b.body` alone would be correct —
 * and it would stop being correct the moment phase 6 adds `body/400/a2/f0`, at
 * which point one slug would match dozens of rows and whichever the engine
 * returned last would become the portrait. Naming the key here means that phase
 * adds rows without changing what a creature page shows.
 */
async function artBySlug() {
  const rows = await query(
    'SELECT b.slug, a.file FROM shard_creature_bodies b ' +
      "JOIN shard_assets a ON a.asset_key = CONCAT('body/', b.body, '/a0') " +
      "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,
  getMeta,
  countAssets,
  replaceBodies,
  allBodies,
  countBodies,
  artBySlug,
  filesForKeys,
  freshKeys,
  countByFamily,
}