tilewright_pack/

read.rs

// Copyright 2026 Seraphina Studios
// SPDX-License-Identifier: MIT

//! Reading and verifying packs. A pack is untrusted input: it is opened
//! read-only and immutable, in SQLite's defensive mode, and its schema must
//! match this version exactly before any other query runs. No value is
//! loaded before SQL has checked its type and size, and verification streams
//! rows instead of collecting whole tables.

use crate::{
    APPLICATION_ID, Error, ErrorCode, FORMAT, META_KEYS, SCHEMA_TABLES, USER_VERSION,
    chunk::{
        CHUNK_ENCODING, ChunkData, DecodeError, FieldRecord, MAXIMUM_CHUNK_BYTES,
        MAXIMUM_RECORD_BYTES, ObjectRecord, PortalRecord, Visit, decode_chunk_with,
    },
    is_field_name, is_identifier, is_namespace,
};
use rusqlite::{
    Connection, OpenFlags, OptionalExtension, Statement, config::DbConfig, params, types,
};
use sha2::{Digest, Sha256};
use std::{
    collections::{BTreeMap, BTreeSet},
    fmt::Write as _,
    io::Read as _,
    path::{Path, PathBuf},
};
use tilewright_jsonc::{Document, Limits, MAX_SAFE_INTEGER, Value};
use tilewright_source::{
    CHUNK_SIZE, MAXIMUM_GROUND_LAYERS, MAXIMUM_LAYER_HEIGHT, Stairwell, StairwellDirection,
    StairwellStep,
};

mod identities;
mod rows;
mod scan;
pub use identities::Identity;
pub use rows::{Anchor, Definition, Page, Plane, Zone};

/// The page size of every pack.
const PAGE_SIZE: i64 = 4096;

/// Largest `meta` value other than `profile`.
const MAXIMUM_META_BYTES: usize = 1024;

/// Largest `sqlite_schema` value loaded; every expected one is shorter.
const MAXIMUM_SCHEMA_BYTES: usize = 1024;

/// The largest `object` index an `object_uids` row may hold. Every object
/// takes more than one byte of a record of at most 16 MiB, so every index
/// is below this.
pub const MAXIMUM_OBJECT_INDEX: usize = MAXIMUM_RECORD_BYTES - 1;

/// The default [`VerifyLimits::maximum_decoded_bytes`]: 8 GiB.
pub const DEFAULT_MAXIMUM_DECODED_BYTES: u64 = 8 * 1024 * 1024 * 1024;

/// The default [`VerifyLimits::maximum_pack_bytes`]: 8 GiB.
pub const DEFAULT_MAXIMUM_PACK_BYTES: u64 = 8 * 1024 * 1024 * 1024;

/// Bounds on the work opening and verifying a pack may do.
#[derive(Clone, Copy, Debug, Eq, PartialEq)]
pub struct VerifyLimits {
    /// Most bytes of chunk records decoded, summed over every chunk.
    /// Verification fails once the chunks' recorded sizes pass it, before
    /// the chunk that would pass it is decompressed.
    pub maximum_decoded_bytes: u64,
    /// Longest pack file opened or hashed. [`Pack::open_with`] checks the
    /// file's length before SQLite opens it, verification checks the pages
    /// again before its integrity check, and hashing fails once it has read
    /// more, even from a file that has grown since it was opened.
    /// Verification uses the lower of this and the limit the pack was
    /// opened with.
    pub maximum_pack_bytes: u64,
}

impl Default for VerifyLimits {
    fn default() -> Self {
        Self {
            maximum_decoded_bytes: DEFAULT_MAXIMUM_DECODED_BYTES,
            maximum_pack_bytes: DEFAULT_MAXIMUM_PACK_BYTES,
        }
    }
}

/// The error for a pack longer than `maximum` bytes.
fn too_long(maximum: u64) -> Error {
    Error::Rejected(
        ErrorCode::TooLarge,
        format!("the pack is more than {maximum} bytes, the pack size limit"),
    )
}

/// Row counts of a pack.
#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)]
pub struct Summary {
    pub strings: u64,
    pub planes: u64,
    pub zones: u64,
    pub anchors: u64,
    pub stairwells: u64,
    pub chunks: u64,
    pub definitions: u64,
    /// Objects in every chunk, inventories included; counted by
    /// [`Pack::verify`] only.
    pub objects: u64,
    /// Rows of `object_uids`: the objects that have a uid.
    pub object_uids: u64,
}

/// Where the object with a uid is: its chunk, and its index in the chunk
/// record's depth-first order (see [`ChunkData::object`]).
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct UidLocation {
    pub plane: String,
    pub layer: i32,
    pub cx: i64,
    pub cy: i64,
    pub index: usize,
}

/// The stairwell step on a tile.
#[derive(Clone, Debug, Eq, PartialEq)]
pub struct StairwellTile {
    /// The stairwell's ID.
    pub stairwell: String,
    /// The step's position in the stairwell's steps.
    pub step: usize,
    pub proto: String,
}

pub struct Pack {
    connection: Connection,
    source: Source,
    /// The pack size limit it was opened with, which bounds
    /// [`Pack::file_digest`].
    maximum_bytes: u64,
}

/// Where a pack's bytes are, which decides what [`Pack::file_digest`]
/// hashes and what verification compares the digest with.
enum Source {
    /// A file, hashed from its path; its name is the expected digest when
    /// it is 64 hex digits.
    File(PathBuf),
    /// An image in memory, hashed from SQLite's read-only copy of it, and
    /// the digest the caller expects, if any.
    Image(Option<[u8; 32]>),
    /// A pack the compiler is checking in memory.
    Compiled,
}

/// Percent-encodes a path for an SQLite URI. On Unix the path's bytes are
/// encoded as they are, so paths that are not UTF-8 open too.
#[cfg_attr(
    unix,
    expect(
        clippy::unnecessary_wraps,
        reason = "only platforms whose paths must be UTF-8 can fail"
    )
)]
fn uri(absolute: &Path) -> Result<String, Error> {
    #[cfg(unix)]
    let bytes = std::os::unix::ffi::OsStrExt::as_bytes(absolute.as_os_str());
    #[cfg(not(unix))]
    let bytes = absolute
        .to_str()
        .ok_or_else(|| Error::Rejected(ErrorCode::Io, "pack path is not UTF-8".into()))?
        .as_bytes();
    // `file://` and then the absolute path, so a path that itself begins with
    // `//` stays part of the file name instead of becoming a URI authority.
    #[cfg(unix)]
    let mut encoded = String::from("file://");
    #[cfg(not(unix))]
    let mut encoded = String::from("file:");
    for byte in bytes {
        if byte.is_ascii_alphanumeric() || b"/-_.~".contains(byte) {
            encoded.push(char::from(*byte));
        } else {
            let _ = write!(encoded, "%{byte:02X}");
        }
    }
    encoded.push_str("?immutable=1");
    Ok(encoded)
}

/// A size bound as an SQL parameter.
fn bound(bytes: usize) -> i64 {
    i64::try_from(bytes).unwrap_or(i64::MAX)
}

/// The largest stored document: the source format's input limit.
fn maximum_document_bytes() -> i64 {
    bound(Limits::default().max_input_bytes)
}

/// The largest interned string: the source format's string limit.
fn maximum_string_bytes() -> i64 {
    bound(Limits::default().max_string_bytes)
}

/// SQL that is true when `column` is not an identifier
/// (`^[a-z0-9][a-z0-9_.-]{0,127}$`). The length is checked first, without
/// loading the value.
fn not_identifier(column: &str) -> String {
    // GLOB stops at a NUL byte, so NUL is rejected separately.
    format!(
        "(typeof({column}) != 'text' OR octet_length({column}) NOT BETWEEN 1 AND 128 \
         OR instr({column}, char(0)) != 0 \
         OR {column} NOT GLOB '[a-z0-9]*' OR {column} GLOB '*[^a-z0-9_.-]*')"
    )
}

/// SQL for a `chunks c` row's data: the blob when the row's encoding is this
/// one and the blob is within [`MAXIMUM_CHUNK_BYTES`], else 0 for another
/// encoding and 1 for other data. Nothing larger is loaded.
fn chunk_data_sql() -> String {
    format!(
        "CASE WHEN octet_length(c.encoding) = {length} AND c.encoding = '{CHUNK_ENCODING}' \
         THEN CASE WHEN typeof(c.data) = 'blob' AND octet_length(c.data) <= {MAXIMUM_CHUNK_BYTES} \
         THEN c.data ELSE 1 END ELSE 0 END",
        length = CHUNK_ENCODING.len()
    )
}

/// The decoded size a chunk's zstd frame records, or 0 when it has none
/// (such a chunk fails to decode).
fn recorded_size(value: &types::Value) -> u64 {
    match value {
        types::Value::Blob(data) => zstd::zstd_safe::get_frame_content_size(data)
            .ok()
            .flatten()
            .unwrap_or(0),
        _ => 0,
    }
}

/// Decodes the value of [`chunk_data_sql`].
fn chunk_data(value: &types::Value) -> Result<ChunkData, Error> {
    chunk_data_with(value, None)
}

/// [`chunk_data`], streaming objects to `visit` when it is given.
fn chunk_data_with(value: &types::Value, visit: Option<Visit<'_>>) -> Result<ChunkData, Error> {
    match value {
        types::Value::Blob(data) => decode_chunk_with(data, visit).map_err(Error::Chunk),
        types::Value::Integer(0) => Err(Error::Rejected(
            ErrorCode::ChunkEncoding,
            format!("a chunk's encoding is not {CHUNK_ENCODING}"),
        )),
        _ => Err(Error::Rejected(
            ErrorCode::ChunkData,
            format!("a chunk's data is not a blob of at most {MAXIMUM_CHUNK_BYTES} bytes"),
        )),
    }
}

/// SQL for `column` when it is text the length of an identifier, at most
/// 128 bytes, else NULL; callers then check the pattern.
fn bounded_identifier(column: &str) -> String {
    format!(
        "CASE WHEN typeof({column}) = 'text' AND octet_length({column}) BETWEEN 1 AND 128 THEN {column} END"
    )
}

/// SQL for `column` when it is an index a source array can have, from 0 to
/// one less than the array limit, else NULL.
fn step_index(column: &str) -> String {
    format!(
        "CASE WHEN typeof({column}) = 'integer' AND {column} BETWEEN 0 AND {} THEN {column} END",
        Limits::default().max_array_items - 1
    )
}

/// SQL for `column` when it is an integer from `minimum` to `maximum`, else
/// NULL.
fn bounded_integer(column: &str, minimum: i64, maximum: i64) -> String {
    format!(
        "CASE WHEN typeof({column}) = 'integer' AND {column} BETWEEN {minimum} AND {maximum} THEN {column} END"
    )
}

/// The chunk coordinates of format 1, whose 32-tile chunks keep every tile
/// within the source format's integer range.
const CHUNKS: (i64, i64) = (
    -(MAX_SAFE_INTEGER / CHUNK_SIZE as i64),
    (MAX_SAFE_INTEGER - (CHUNK_SIZE as i64 - 1)) / CHUNK_SIZE as i64,
);

/// The columns of an `object_uids` row, each NULL when out of bounds.
type StoredLocation = (
    Option<String>,
    Option<i64>,
    Option<i64>,
    Option<i64>,
    Option<i64>,
);

/// The error for an `object_uids` row that is not one a compiler writes.
fn bad_uid_row() -> Error {
    Error::Rejected(
        ErrorCode::ObjectUid,
        "an object_uids row's plane is not an identifier, or its layer, chunk or index is out of range".into(),
    )
}

/// SQL for `column` when it is text of at most `?1` bytes, else NULL.
fn bounded_text(column: &str) -> String {
    format!(
        "CASE WHEN typeof({column}) = 'text' AND octet_length({column}) <= ?1 THEN {column} END"
    )
}

/// Parses a stored document, which must be canonical JSON of an object;
/// `code` is the rule a bad one breaks.
fn stored_document(text: Option<String>, what: &str, code: ErrorCode) -> Result<Document, Error> {
    let invalid = || {
        Error::Rejected(
            code,
            format!("{what} is not canonical JSON of an object with namespaced custom data"),
        )
    };
    let text = text.ok_or_else(invalid)?;
    if !scan::compact(&text) {
        return Err(invalid());
    }
    let document = Document::parse(text, &crate::canonical_limits()).map_err(|_| invalid())?;
    // `custom` data, where a document has it, is keyed by namespaces as in
    // the source format.
    let custom_ok = match document.root().get("custom") {
        None => true,
        Some(custom) => custom
            .as_object()
            .is_some_and(|members| members.iter().all(|member| is_namespace(&member.key))),
    };
    if document.root().as_object().is_none()
        || document.canonical(document.root()) != document.source()
        || !custom_ok
    {
        return Err(invalid());
    }
    Ok(document)
}

/// Whether a stored profile's `projection` is one a source profile can
/// have: `isometric` with `tile_width` and `tile_height`, or `top_down` with
/// `tile_size`, each an integer from 1 to `u32::MAX`, and no other keys.
fn projection_ok(profile: &Document) -> bool {
    let Some(members) = profile.root().get("projection").and_then(Value::as_object) else {
        return false;
    };
    let keys: BTreeSet<&str> = members.iter().map(|member| member.key.as_str()).collect();
    let sizes: &[&str] = match members
        .iter()
        .find(|member| member.key == "kind")
        .and_then(|member| member.value.as_str())
    {
        Some("isometric") => &["tile_width", "tile_height"],
        Some("top_down") => &["tile_size"],
        _ => return false,
    };
    keys.len() == sizes.len() + 1
        && sizes.iter().all(|size| {
            members
                .iter()
                .find(|member| member.key == *size)
                .and_then(|member| member.value.as_i64())
                .is_some_and(|value| (1..=i64::from(u32::MAX)).contains(&value))
        })
}

/// The `layer_height` of a stored profile: `Ok(None)` when it has none, and
/// an error unless it is an integer from 1 to [`MAXIMUM_LAYER_HEIGHT`].
fn stored_layer_height(profile: &Document) -> Result<Option<u32>, Error> {
    profile
        .root()
        .get("layer_height")
        .map(|value| {
            value
                .as_i64()
                .filter(|height| (1..=i64::from(MAXIMUM_LAYER_HEIGHT)).contains(height))
                .and_then(|height| u32::try_from(height).ok())
                .ok_or_else(|| {
                    Error::Rejected(
                        ErrorCode::Profile,
                        format!(
                            "the stored profile's layer_height is not an integer from 1 to {MAXIMUM_LAYER_HEIGHT}"
                        ),
                    )
                })
        })
        .transpose()
}

/// Reads a stored stairwell document, already checked to be canonical JSON
/// of an object with namespaced custom data, into a stairwell with the
/// source format's shape: an identifier, at least two connected steps,
/// each a 32-bit layer, a position within the integer range and a
/// prototype identifier, and a valid direction and flag.
fn stored_stairwell(document: &Document) -> Result<Stairwell, Error> {
    let invalid = || {
        Error::Rejected(
            ErrorCode::Stairwell,
            "a stairwell document does not have the source format's shape".into(),
        )
    };
    let closed = |value: &Value, keys: &[&str]| {
        value.as_object().is_some_and(|members| {
            members
                .iter()
                .all(|member| keys.contains(&member.key.as_str()))
        })
    };
    let root = document.root();
    if !closed(
        root,
        &[
            "id",
            "steps",
            "direction",
            "hidden_on_other_layers",
            "custom",
        ],
    ) {
        return Err(invalid());
    }
    let id = text(document, "id")
        .filter(|id| is_identifier(id))
        .ok_or_else(invalid)?;
    let direction = match root.get("direction") {
        None => StairwellDirection::Both,
        Some(value) => value
            .as_str()
            .and_then(StairwellDirection::parse)
            .ok_or_else(invalid)?,
    };
    let hidden_on_other_layers = match root.get("hidden_on_other_layers") {
        None => true,
        Some(value) => value.as_bool().ok_or_else(invalid)?,
    };
    let in_range = |value: &Value| {
        value
            .as_i64()
            .filter(|number| (-MAX_SAFE_INTEGER..=MAX_SAFE_INTEGER).contains(number))
    };
    let mut steps = Vec::new();
    for step in root
        .get("steps")
        .and_then(Value::as_array)
        .ok_or_else(invalid)?
    {
        if !closed(step, &["layer", "at", "proto"]) {
            return Err(invalid());
        }
        let layer = step
            .get("layer")
            .and_then(Value::as_i64)
            .and_then(|layer| i32::try_from(layer).ok());
        let at = match step.get("at").and_then(Value::as_array) {
            Some([x, y]) => in_range(x).zip(in_range(y)).map(|(x, y)| [x, y]),
            _ => None,
        };
        let proto = step
            .get("proto")
            .and_then(Value::as_str)
            .filter(|proto| is_identifier(proto));
        let (Some(layer), Some(at), Some(proto)) = (layer, at, proto) else {
            return Err(invalid());
        };
        steps.push(StairwellStep {
            layer,
            at,
            proto: proto.to_owned(),
        });
    }
    if !Stairwell::connects(&steps) {
        return Err(Error::Rejected(
            ErrorCode::Stairwell,
            format!(
                "the steps of stairwell {id} are not on adjacent layers in one direction, at most one tile apart"
            ),
        ));
    }
    Ok(Stairwell {
        id: id.to_owned(),
        steps,
        direction,
        hidden_on_other_layers,
        custom: root.get("custom").map(|custom| document.canonical(custom)),
        canonical: document.source().to_owned(),
    })
}

/// A string member of a document's root object.
fn text<'d>(document: &'d Document, key: &str) -> Option<&'d str> {
    document.root().get(key).and_then(Value::as_str)
}

/// Column `index` as text: `None` when it is not text, as a bounded column
/// is when out of bounds. SQLite stores any bytes as text, so text that is
/// not UTF-8 is `code`'s error rather than an SQLite one.
fn utf8_column(
    row: &rusqlite::Row<'_>,
    index: usize,
    code: ErrorCode,
) -> Result<Option<String>, Error> {
    match row.get_ref(index)? {
        types::ValueRef::Text(bytes) => std::str::from_utf8(bytes)
            .map(|text| Some(text.to_owned()))
            .map_err(|_| Error::Rejected(code, "a stored value is not UTF-8 text".into())),
        _ => Ok(None),
    }
}

/// A stairwell from a row of `id, document`, bounded in SQL by
/// [`bounded_identifier`] and [`bounded_text`] (NULL when out of bounds),
/// after checking that the document has the row's ID. It fails with
/// [`ErrorCode::DecodeBudget`] when the document lists more than
/// `maximum_steps` steps, before it is parsed.
pub(super) fn stairwell_row_within(
    row: &rusqlite::Row<'_>,
    maximum_steps: u64,
) -> Result<Stairwell, Error> {
    let id = utf8_column(row, 0, ErrorCode::Stairwell)?;
    let text = utf8_column(row, 1, ErrorCode::Stairwell)?;
    // The steps are counted before the document is parsed: a scan of the
    // text, exact for the canonical JSON a stored document must be.
    let steps = text
        .as_deref()
        .filter(|text| scan::compact(text))
        .and_then(|text| scan::root_array_items(text, "steps"));
    if steps.is_some_and(|steps| u64::try_from(steps).unwrap_or(u64::MAX) > maximum_steps) {
        return Err(Error::Rejected(
            ErrorCode::DecodeBudget,
            format!("a stairwell lists more than {maximum_steps} steps, the read's budget"),
        ));
    }
    let document = stored_document(text, "a stairwell document", ErrorCode::Stairwell)?;
    let stairwell = stored_stairwell(&document)?;
    if id.as_deref() != Some(stairwell.id.as_str()) {
        return Err(Error::Rejected(
            ErrorCode::Stairwell,
            format!(
                "the document of stairwell {} disagrees with its row",
                stairwell.id
            ),
        ));
    }
    Ok(stairwell)
}

impl Pack {
    /// Opens a pack with [`VerifyLimits::default`]; see [`Pack::open_with`].
    ///
    /// # Errors
    /// Returns an error if the file is not a pack of this format or is
    /// longer than the default pack size limit.
    pub fn open(path: &Path) -> Result<Self, Error> {
        Self::open_with(path, &VerifyLimits::default())
    }

    /// Opens a pack and checks its size, identity and schema. A file longer
    /// than `limits.maximum_pack_bytes` is rejected before SQLite opens it,
    /// and [`Pack::file_digest`] reads at most that many bytes.
    ///
    /// # Errors
    /// Returns an error if the file is not a pack of this format or is
    /// longer than the limit.
    pub fn open_with(path: &Path, limits: &VerifyLimits) -> Result<Self, Error> {
        // The same absolute path opens the database and is hashed later, so
        // a change of working directory cannot point them at different files.
        let path = std::path::absolute(path)?;
        let metadata = std::fs::metadata(&path)?;
        if !metadata.is_file() {
            return Err(Error::Rejected(
                ErrorCode::NotAFile,
                format!("{} is not a file", path.display()),
            ));
        }
        if metadata.len() > limits.maximum_pack_bytes {
            return Err(too_long(limits.maximum_pack_bytes));
        }
        let mut head = Vec::with_capacity(HEADER_BYTES);
        std::fs::File::open(&path)?
            .take(HEADER_BYTES as u64)
            .read_to_end(&mut head)?;
        check_header(&head)?;
        let connection = Connection::open_with_flags(
            uri(&path)?,
            OpenFlags::SQLITE_OPEN_READ_ONLY
                | OpenFlags::SQLITE_OPEN_URI
                | OpenFlags::SQLITE_OPEN_NO_MUTEX,
        )?;
        harden(&connection)?;
        check_identity(&connection)?;
        // The file SQLite opened may have replaced the one whose length was
        // checked, so its pages are checked against the limit too.
        check_pages(&connection, std::fs::metadata(&path)?.len(), limits)?;
        Ok(Self {
            connection,
            source: Source::File(path),
            maximum_bytes: limits.maximum_pack_bytes,
        })
    }

    /// Opens a pack from its bytes, as [`Pack::open_from`] does.
    ///
    /// # Errors
    /// Returns an error if the bytes are not a pack of this format or are
    /// more than the pack size limit.
    pub fn open_bytes(
        bytes: &[u8],
        digest: Option<[u8; 32]>,
        limits: &VerifyLimits,
    ) -> Result<Self, Error> {
        let length = u64::try_from(bytes.len()).unwrap_or(u64::MAX);
        Self::open_from(bytes, length, digest, limits)
    }

    /// Opens a pack image of exactly `length` bytes, which `read` yields, and checks
    /// its size, identity and schema as [`Pack::open_with`] checks a file's.
    /// An image longer than `limits.maximum_pack_bytes` is rejected before
    /// any of it is read. SQLite reads a read-only copy in memory, which
    /// stays the pack's bytes until it is dropped.
    ///
    /// `digest`, when given, is the SHA-256 the image must have. Like a file
    /// name that is a digest, verification compares it with the image's
    /// digest; opening does not hash the image.
    ///
    /// # Errors
    /// Returns an error if `read` fails, ends early or yields more than
    /// `length` bytes, the image is longer than the limit or than memory can
    /// hold, or it is not a pack of this format.
    pub fn open_from(
        read: impl std::io::Read,
        length: u64,
        digest: Option<[u8; 32]>,
        limits: &VerifyLimits,
    ) -> Result<Self, Error> {
        if length > limits.maximum_pack_bytes {
            return Err(too_long(limits.maximum_pack_bytes));
        }
        let size = usize::try_from(length).map_err(|_| {
            Error::Io(std::io::Error::new(
                std::io::ErrorKind::OutOfMemory,
                "the pack is larger than memory can hold",
            ))
        })?;
        let mut connection = Connection::open_in_memory_with_flags(
            OpenFlags::SQLITE_OPEN_READ_ONLY | OpenFlags::SQLITE_OPEN_NO_MUTEX,
        )?;
        let mut image = Filled {
            inner: read,
            missing: false,
            failure: None,
            head: Vec::with_capacity(HEADER_BYTES),
        };
        // An empty image is an empty database, as an empty file is; SQLite
        // cannot allocate a copy of no bytes.
        if size > 0 {
            // `Filled` never fails, so SQLite always takes ownership of its
            // copy and frees it with the connection, even when the image
            // could not be read.
            connection.deserialize_read_exact(rusqlite::MAIN_DB, &mut image, size, true)?;
        }
        if let Some(error) = image.failure {
            return Err(Error::Io(error));
        }
        if image.missing {
            return Err(Error::Io(std::io::Error::new(
                std::io::ErrorKind::UnexpectedEof,
                format!("the pack image ends before its {length} bytes"),
            )));
        }
        // The image is exactly `length` bytes: one that goes on is not the
        // pack its length describes.
        if read_some(&mut image.inner)? {
            return Err(Error::Rejected(
                ErrorCode::FileLength,
                format!("the pack image is longer than {length} bytes"),
            ));
        }
        // SQLite has not read the copy yet: it first does so in
        // `check_identity`.
        check_header(&image.head)?;
        harden(&connection)?;
        check_identity(&connection)?;
        check_pages(&connection, length, limits)?;
        Ok(Self {
            connection,
            source: Source::Image(digest),
            maximum_bytes: limits.maximum_pack_bytes,
        })
    }

    /// A pack the compiler has just built, checked like an opened one.
    pub(crate) fn compiled(connection: Connection) -> Result<Self, Error> {
        check_identity(&connection)?;
        Ok(Self {
            connection,
            source: Source::Compiled,
            // There is no file to hash, so the limit is never used.
            maximum_bytes: u64::MAX,
        })
    }

    pub(crate) fn into_connection(self) -> Connection {
        self.connection
    }

    /// A `meta` value.
    ///
    /// # Errors
    /// Returns an error if SQLite fails.
    pub fn meta(&self, key: &str) -> Result<Option<String>, Error> {
        let mut statement = self
            .connection
            .prepare("SELECT value FROM meta WHERE key = ?1")?;
        let mut rows = statement.query([key])?;
        match rows.next()? {
            Some(row) => utf8_column(row, 0, ErrorCode::Meta),
            None => Ok(None),
        }
    }

    /// An interned string.
    ///
    /// # Errors
    /// Returns an error if SQLite fails or the stored value is not text of
    /// at most 1 MiB.
    pub fn string(&self, id: u32) -> Result<Option<String>, Error> {
        let mut statement = self.connection.prepare(&format!(
            "SELECT {} FROM strings WHERE id = ?2",
            bounded_text("value")
        ))?;
        let mut rows = statement.query(params![maximum_string_bytes(), id])?;
        let Some(row) = rows.next()? else {
            return Ok(None);
        };
        match utf8_column(row, 0, ErrorCode::String)? {
            Some(text) => Ok(Some(text)),
            None => Err(Error::Rejected(
                ErrorCode::String,
                "a string is not text of at most 1 MiB".into(),
            )),
        }
    }

    /// Decodes one chunk into its full model; see [`decode_chunk`].
    ///
    /// [`decode_chunk`]: crate::decode_chunk
    ///
    /// # Errors
    /// Returns an error if SQLite fails or the stored record is invalid.
    pub fn chunk(
        &self,
        plane: &str,
        layer: i32,
        cx: i64,
        cy: i64,
    ) -> Result<Option<ChunkData>, Error> {
        self.chunk_within(plane, layer, cx, cy, u64::MAX)
    }

    /// [`Pack::chunk`], with a budget of `maximum_entries` entries: its
    /// objects, inventories included, and their fields. A record of at most
    /// 16 MiB can hold about a million small objects, or as many fields, so
    /// a caller that builds a value for each, such as a browser, can bound
    /// them. Under a budget the record is first decoded keeping one object
    /// at a time, and the read fails with [`ErrorCode::DecodeBudget`] once
    /// it counts more, before any chunk is built.
    ///
    /// # Errors
    /// Returns an error if SQLite fails, the stored record is invalid, or it
    /// holds more entries than the budget.
    pub fn chunk_within(
        &self,
        plane: &str,
        layer: i32,
        cx: i64,
        cy: i64,
        maximum_entries: u64,
    ) -> Result<Option<ChunkData>, Error> {
        let value: Option<types::Value> = self
            .connection
            .query_row(
                &format!(
                    "SELECT {} FROM chunks c WHERE c.plane = ?1 AND c.layer = ?2 AND c.cx = ?3 AND c.cy = ?4",
                    chunk_data_sql()
                ),
                params![plane, layer, cx, cy],
                |row| row.get(0),
            )
            .optional()?;
        let Some(value) = value else {
            return Ok(None);
        };
        if maximum_entries < u64::MAX {
            let mut entries: u64 = 0;
            let mut over = false;
            let mut count = |object: &ObjectRecord| {
                let fields = u64::try_from(object.fields.len()).unwrap_or(u64::MAX);
                entries = entries.saturating_add(1).saturating_add(fields);
                over = entries > maximum_entries;
                if over {
                    Err(DecodeError(
                        "the chunk holds more entries than the budget".into(),
                    ))
                } else {
                    Ok(())
                }
            };
            let counted = chunk_data_with(&value, Some(&mut count));
            if over {
                return Err(Error::Rejected(
                    ErrorCode::DecodeBudget,
                    format!(
                        "the chunk holds more than {maximum_entries} objects and fields, the read's budget"
                    ),
                ));
            }
            counted?;
        }
        chunk_data(&value).map(Some)
    }

    /// The profile's `layer_height`: height units per layer, or `None` when
    /// the profile does not stack layers. The stored profile is checked as a
    /// full verification checks it.
    ///
    /// # Errors
    /// Returns an error if SQLite fails or the stored profile is invalid.
    pub fn layer_height(&self) -> Result<Option<u32>, Error> {
        Ok(self.profile_layers()?.2)
    }

    /// One stairwell of a plane, read from its stored document. Like
    /// [`Pack::chunk`], it checks the shape and bounds of what it loads;
    /// whether the plane and the steps' prototypes exist is checked by
    /// [`Pack::verify`].
    ///
    /// # Errors
    /// Returns an error if SQLite fails, or the stored stairwell does not
    /// have the source format's shape or disagrees with its row.
    pub fn stairwell(&self, plane: &str, id: &str) -> Result<Option<Stairwell>, Error> {
        self.stairwell_within(plane, id, u64::MAX)
    }

    /// [`Pack::stairwell`], with a budget of `maximum_steps` steps. A stored
    /// document can list about a quarter of a million, so a caller that
    /// builds a value for each, such as a browser, can bound them; the read
    /// fails with [`ErrorCode::DecodeBudget`] before it builds any step.
    ///
    /// # Errors
    /// Returns an error if SQLite fails, the stored stairwell is invalid or
    /// disagrees with its row, or it lists more steps than the budget.
    pub fn stairwell_within(
        &self,
        plane: &str,
        id: &str,
        maximum_steps: u64,
    ) -> Result<Option<Stairwell>, Error> {
        let mut statement = self.connection.prepare(&format!(
            "SELECT {}, {} FROM stairwells WHERE plane = ?2 AND id = ?3",
            bounded_identifier("id"),
            bounded_text("document"),
        ))?;
        let mut rows = statement.query(params![maximum_document_bytes(), plane, id])?;
        rows.next()?
            .map(|row| stairwell_row_within(row, maximum_steps))
            .transpose()
    }

    /// The stairwell step on a tile of a plane, if any. It checks the shape
    /// and bounds of what it loads; whether the stairwell and prototype
    /// exist is checked by [`Pack::verify`].
    ///
    /// # Errors
    /// Returns an error if SQLite fails or the stored step is invalid.
    pub fn stairwell_step(
        &self,
        plane: &str,
        layer: i32,
        x: i64,
        y: i64,
    ) -> Result<Option<StairwellTile>, Error> {
        let mut statement = self.connection.prepare(&format!(
            "SELECT {}, {}, {} FROM stairwell_steps \
             WHERE plane = ?1 AND layer = ?2 AND x = ?3 AND y = ?4",
            bounded_identifier("stairwell"),
            step_index("step"),
            bounded_identifier("proto"),
        ))?;
        let mut rows = statement.query(params![plane, layer, x, y])?;
        let Some(row) = rows.next()? else {
            return Ok(None);
        };
        let found: (Option<String>, Option<i64>, Option<String>) = (
            utf8_column(row, 0, ErrorCode::StairwellStep)?,
            row.get(1)?,
            utf8_column(row, 2, ErrorCode::StairwellStep)?,
        );
        match found {
            (Some(stairwell), Some(step), Some(proto))
                if is_identifier(&stairwell) && is_identifier(&proto) =>
            {
                Ok(Some(StairwellTile {
                    stairwell,
                    step: usize::try_from(step).map_err(|_| {
                        Error::Rejected(
                            ErrorCode::StairwellStep,
                            "a stairwell step has a negative index".into(),
                        )
                    })?,
                    proto,
                }))
            }
            _ => Err(Error::Rejected(
                ErrorCode::StairwellStep,
                "a stairwell step's stairwell or proto is not an identifier, or its index is not one a source array can have".into(),
            )),
        }
    }

    /// Where the object with `uid` is, from its `object_uids` row. Like
    /// [`Pack::chunk`], it checks the shape and bounds of what it loads;
    /// whether the row agrees with the chunk records is checked by
    /// [`Pack::verify`], or for this one object by [`Pack::object`].
    ///
    /// # Errors
    /// Returns an error if SQLite fails or the stored row is invalid.
    pub fn object_location(&self, uid: &str) -> Result<Option<UidLocation>, Error> {
        if !is_identifier(uid) {
            return Ok(None);
        }
        let mut statement = self.connection.prepare(&format!(
            "SELECT {}, {}, {}, {}, {} FROM object_uids WHERE uid = ?1",
            bounded_identifier("plane"),
            bounded_integer("layer", i64::from(i32::MIN), i64::from(i32::MAX)),
            bounded_integer("cx", CHUNKS.0, CHUNKS.1),
            bounded_integer("cy", CHUNKS.0, CHUNKS.1),
            bounded_integer("object", 0, bound(MAXIMUM_OBJECT_INDEX)),
        ))?;
        let mut rows = statement.query([uid])?;
        let found: Option<StoredLocation> = match rows.next()? {
            Some(row) => Some((
                utf8_column(row, 0, ErrorCode::ObjectUid)?,
                row.get(1)?,
                row.get(2)?,
                row.get(3)?,
                row.get(4)?,
            )),
            None => None,
        };
        let Some(found) = found else {
            return Ok(None);
        };
        match found {
            (Some(plane), Some(layer), Some(cx), Some(cy), Some(index))
                if is_identifier(&plane) =>
            {
                Ok(Some(UidLocation {
                    plane,
                    layer: i32::try_from(layer).map_err(|_| bad_uid_row())?,
                    cx,
                    cy,
                    index: usize::try_from(index).map_err(|_| bad_uid_row())?,
                }))
            }
            _ => Err(bad_uid_row()),
        }
    }

    /// The object with `uid` and where it is. It decodes the object's chunk,
    /// as [`Pack::chunk`] does, and fails unless the object at the row's
    /// index has `uid`.
    ///
    /// # Errors
    /// Returns an error if SQLite fails, the stored row or chunk is invalid,
    /// or they disagree.
    pub fn object(&self, uid: &str) -> Result<Option<(UidLocation, ObjectRecord)>, Error> {
        self.object_within(uid, u64::MAX)
    }

    /// [`Pack::object`], decoding the object's chunk with a budget of
    /// `maximum_entries` objects and fields; see [`Pack::chunk_within`].
    ///
    /// # Errors
    /// Returns an error if SQLite fails, the stored row or chunk is invalid,
    /// they disagree, or the chunk holds more entries than the budget.
    pub fn object_within(
        &self,
        uid: &str,
        maximum_entries: u64,
    ) -> Result<Option<(UidLocation, ObjectRecord)>, Error> {
        let Some(location) = self.object_location(uid)? else {
            return Ok(None);
        };
        let object = self
            .chunk_within(
                &location.plane,
                location.layer,
                location.cx,
                location.cy,
                maximum_entries,
            )?
            .and_then(|chunk| {
                chunk
                    .object(location.index)
                    .filter(|object| object.uid.as_deref() == Some(uid))
                    .cloned()
            })
            .ok_or_else(|| {
                Error::Rejected(
                    ErrorCode::ObjectUid,
                    format!("the object_uids row of {uid} does not name an object with that uid"),
                )
            })?;
        Ok(Some((location, object)))
    }

    /// Row counts.
    ///
    /// # Errors
    /// Returns an error if SQLite fails.
    pub fn summary(&self) -> Result<Summary, Error> {
        let count = |table: &str| -> Result<u64, Error> {
            let value: i64 =
                self.connection
                    .query_row(&format!("SELECT count(*) FROM {table}"), [], |row| {
                        row.get(0)
                    })?;
            Ok(u64::try_from(value).unwrap_or(0))
        };
        Ok(Summary {
            strings: count("strings")?,
            planes: count("planes")?,
            zones: count("zones")?,
            anchors: count("anchors")?,
            stairwells: count("stairwells")?,
            chunks: count("chunks")?,
            definitions: count("definitions")?,
            objects: 0,
            object_uids: count("object_uids")?,
        })
    }

    /// The SHA-256 of the pack file, read from its path when called, or of
    /// the image a pack was opened from.
    ///
    /// For a file, this hashes whatever file is at the pack's path now, not
    /// necessarily the file SQLite opened: the digest binds to the path.
    /// Packs are content-addressed files opened from directories the caller
    /// controls; a directory that others can write to is not supported. It
    /// reads at most the pack size limit the pack was opened with, and fails
    /// for a longer file. An image is hashed from the read-only copy SQLite
    /// reads, so its digest binds to the bytes the reads see.
    ///
    /// # Errors
    /// Returns an error if the file cannot be read or is longer than the
    /// limit.
    pub fn file_digest(&self) -> Result<[u8; 32], Error> {
        self.digest_within(self.maximum_bytes)
    }

    /// [`Pack::file_digest`], reading at most `maximum` bytes.
    fn digest_within(&self, maximum: u64) -> Result<[u8; 32], Error> {
        let path = match &self.source {
            Source::File(path) => path,
            Source::Image(_) => {
                let image = self.connection.serialize(rusqlite::MAIN_DB)?;
                if u64::try_from(image.len()).map_or(true, |length| length > maximum) {
                    return Err(too_long(maximum));
                }
                return Ok(Sha256::digest(&*image).into());
            }
            Source::Compiled => {
                return Err(Error::Rejected(
                    ErrorCode::NotAFile,
                    "the pack is not a file".into(),
                ));
            }
        };
        let file = std::fs::File::open(path)?;
        if file.metadata()?.len() > maximum {
            return Err(too_long(maximum));
        }
        // The length can change once checked, so the read stops one byte
        // past the limit.
        let mut file = file.take(maximum.saturating_add(1));
        let mut hasher = Sha256::new();
        let mut buffer = vec![0; 64 * 1024];
        let mut total: u64 = 0;
        loop {
            let read = file.read(&mut buffer)?;
            if read == 0 {
                break;
            }
            total = total.saturating_add(u64::try_from(read).unwrap_or(u64::MAX));
            if total > maximum {
                return Err(too_long(maximum));
            }
            hasher.update(&buffer[..read]);
        }
        Ok(hasher.finalize().into())
    }

    /// Checks the whole pack with the default decoded-byte budget and the
    /// pack size limit the pack was opened with; see [`Pack::verify_with`].
    ///
    /// # Errors
    /// Returns the first problem found.
    pub fn verify(&self) -> Result<Summary, Error> {
        self.verify_with(&VerifyLimits {
            maximum_pack_bytes: self.maximum_bytes,
            ..VerifyLimits::default()
        })
    }

    /// Checks the whole pack: SQLite integrity, the file name digest when the
    /// name is one, string, plane, zone, anchor and definition rows, every
    /// stored document, every chunk record, and every string, layer, ground
    /// count and portal target a record references.
    ///
    /// Decoding every chunk is linear in the pack's decoded size, which
    /// compression can make far larger than the file, so it stops and fails
    /// once `limits.maximum_decoded_bytes` would be passed. The integrity
    /// check and the hash are linear in the file's size, so the pages are
    /// checked against the pack size limit before either runs, and the hash
    /// reads at most that many bytes. The limit is the lower of
    /// `limits.maximum_pack_bytes` and the one the pack was opened with, so
    /// verifying never reads more than opening allowed. The file name check
    /// hashes the file at the pack's path, so it binds to the path; see
    /// [`Pack::file_digest`].
    ///
    /// # Errors
    /// Returns the first problem found, including an exceeded limit.
    pub fn verify_with(&self, limits: &VerifyLimits) -> Result<Summary, Error> {
        let maximum_bytes = limits.maximum_pack_bytes.min(self.maximum_bytes);
        let pages: i64 = self
            .connection
            .query_row("PRAGMA page_count", [], |row| row.get(0))?;
        if u64::try_from(pages)
            .ok()
            .and_then(|pages| pages.checked_mul(PAGE_SIZE.unsigned_abs()))
            .is_none_or(|bytes| bytes > maximum_bytes)
        {
            return Err(too_long(maximum_bytes));
        }
        let integrity: String = self
            .connection
            .query_row("PRAGMA integrity_check", [], |row| row.get(0))?;
        if integrity != "ok" {
            return Err(Error::Rejected(
                ErrorCode::Integrity,
                format!("integrity check failed: {integrity}"),
            ));
        }
        if let Some(expected) = self.expected_digest() {
            // A name must be the lowercase digest; other cases do not match.
            let digest = crate::compile::hex(&self.digest_within(maximum_bytes)?);
            if digest != expected {
                return Err(Error::Rejected(
                    ErrorCode::Digest,
                    match self.source {
                        Source::File(_) => format!("file digest {digest} does not match its name"),
                        _ => format!("pack digest {digest} is not {expected}, the expected digest"),
                    },
                ));
            }
        }
        self.verify_contents(limits.maximum_decoded_bytes)
    }

    /// The digest verification expects, in hex: a file's name when it is 64
    /// hex digits, or the digest an image was opened with.
    fn expected_digest(&self) -> Option<String> {
        match &self.source {
            Source::File(path) => path
                .file_name()
                .and_then(|name| name.to_str())
                .and_then(|name| name.strip_suffix(".twpack"))
                .filter(|stem| {
                    stem.len() == 64 && stem.bytes().all(|byte| byte.is_ascii_hexdigit())
                })
                .map(str::to_owned),
            Source::Image(digest) => digest.as_ref().map(|digest| crate::compile::hex(digest)),
            Source::Compiled => None,
        }
    }

    /// Everything [`Pack::verify`] checks except SQLite integrity and the
    /// file name: what the compiler checks of every pack it builds.
    pub(crate) fn verify_contents(&self, maximum_decoded_bytes: u64) -> Result<Summary, Error> {
        let mut summary = self.summary()?;
        let (minimum, maximum): (Option<i64>, Option<i64>) =
            self.connection
                .query_row("SELECT min(id), max(id) FROM strings", [], |row| {
                    Ok((row.get(0)?, row.get(1)?))
                })?;
        let contiguous = match (minimum, maximum) {
            (None, None) => summary.strings == 0,
            (Some(1), Some(maximum)) => u64::try_from(maximum).ok() == Some(summary.strings),
            _ => false,
        };
        if !contiguous {
            return Err(Error::Rejected(
                ErrorCode::StringIds,
                "string IDs are not 1..=count".into(),
            ));
        }
        self.check_rows()?;
        self.check_zone_trees()?;
        self.check_string_encoding()?;
        self.check_documents()?;
        self.check_stairwells()?;
        let (render_layers, ground_layers, _) = self.profile_layers()?;
        let strings = u32::try_from(summary.strings)
            .map_err(|_| Error::Rejected(ErrorCode::StringIds, "too many strings".into()))?;
        let mut targets = Targets::new(&self.connection, strings, render_layers)?;
        let mut statement = self.connection.prepare(&format!(
            "SELECT p.id, p.chunk_size, c.layer, c.cx, c.cy, {} FROM chunks c JOIN planes p ON p.id = c.plane ORDER BY c.plane, c.layer, c.cx, c.cy",
            chunk_data_sql()
        ))?;
        let mut rows = statement.query([])?;
        let mut checked = 0;
        let mut decoded: u64 = 0;
        while let Some(row) = rows.next()? {
            let plane: String = row.get(0)?;
            let chunk_size: u16 = row.get(1)?;
            let key: (i32, i64, i64) = (row.get(2)?, row.get(3)?, row.get(4)?);
            let data: types::Value = row.get(5)?;
            // The frame records its decoded size, so the budget is checked
            // before the chunk is decompressed.
            decoded = decoded.saturating_add(recorded_size(&data));
            if decoded > maximum_decoded_bytes {
                return Err(Error::Rejected(
                    ErrorCode::DecodeBudget,
                    format!(
                        "the chunks decode to more than {maximum_decoded_bytes} bytes, the verification limit"
                    ),
                ));
            }
            // Objects are checked as they decode rather than kept, so memory
            // stays bounded by the record, not by the objects it holds.
            targets.enter(&plane, key);
            let mut failure = None;
            let mut visit = |object: &ObjectRecord| {
                summary.objects += 1;
                targets.check(object).map_err(|error| {
                    failure = Some(error);
                    DecodeError("an object failed verification".into())
                })
            };
            let decoded_chunk = chunk_data_with(&data, Some(&mut visit));
            if let Some(error) = failure {
                return Err(error);
            }
            let chunk = decoded_chunk?;
            if chunk.chunk_size != chunk_size || chunk.ground.len() != ground_layers {
                return Err(Error::Rejected(
                    ErrorCode::Chunk,
                    "a chunk disagrees with its plane or profile".into(),
                ));
            }
            targets.check_palette(&chunk)?;
            checked += 1;
        }
        if checked != summary.chunks {
            return Err(Error::Rejected(
                ErrorCode::Chunk,
                "a chunk names an unknown plane".into(),
            ));
        }
        // Each object with a uid matched the row of its uid, which names
        // only that object, so the rows the objects matched are distinct;
        // with as many rows as such objects, no row is left over and no two
        // objects share a uid.
        if targets.uids != summary.object_uids {
            return Err(Error::Rejected(
                ErrorCode::ObjectUid,
                "object_uids holds a uid that no object has".into(),
            ));
        }
        Ok(summary)
    }

    /// Checks string, plane, zone, anchor and definition rows on their own
    /// and against each other, without loading them.
    fn check_rows(&self) -> Result<(), Error> {
        let checks = [
            (
                format!(
                    "SELECT count(*) FROM strings WHERE typeof(value) != 'text' OR octet_length(value) > {} OR instr(value, char(0)) != 0",
                    maximum_string_bytes()
                ),
                ErrorCode::String,
                "a string is not text of at most 1 MiB without NUL",
            ),
            (
                "SELECT count(*) FROM strings a JOIN strings b ON b.id = a.id + 1 WHERE b.value <= a.value".to_owned(),
                ErrorCode::StringIds,
                "string IDs are not in byte order of their values",
            ),
            (
                format!(
                    "SELECT count(*) FROM planes WHERE {} OR kind NOT IN ('shared', 'instanced', 'per_player') OR typeof(chunk_size) != 'integer' OR chunk_size != {CHUNK_SIZE}",
                    not_identifier("id")
                ),
                ErrorCode::Plane,
                "a plane has an invalid ID or kind, or a chunk size other than 32",
            ),
            (
                format!(
                    "SELECT count(*) FROM zones z WHERE {} OR NOT EXISTS (SELECT 1 FROM planes p WHERE p.id = z.plane) OR (z.parent IS NOT NULL AND NOT EXISTS (SELECT 1 FROM zones q WHERE q.plane = z.plane AND q.id = z.parent))",
                    not_identifier("z.id")
                ),
                ErrorCode::Zone,
                "a zone has an invalid ID or names an unknown plane or parent",
            ),
            (
                format!(
                    "SELECT count(*) FROM anchors a WHERE {} OR NOT EXISTS (SELECT 1 FROM planes p WHERE p.id = a.plane) OR typeof(a.layer) != 'integer' OR a.layer NOT BETWEEN -2147483648 AND 2147483647 OR typeof(a.x) != 'integer' OR a.x NOT BETWEEN -{MAX_SAFE_INTEGER} AND {MAX_SAFE_INTEGER} OR typeof(a.y) != 'integer' OR a.y NOT BETWEEN -{MAX_SAFE_INTEGER} AND {MAX_SAFE_INTEGER}",
                    not_identifier("a.id")
                ),
                ErrorCode::Anchor,
                "an anchor has an invalid ID, layer or position, or names an unknown plane",
            ),
            (
                format!(
                    "SELECT count(*) FROM stairwells s WHERE {} OR NOT EXISTS (SELECT 1 FROM planes p WHERE p.id = s.plane)",
                    not_identifier("s.id")
                ),
                ErrorCode::Stairwell,
                "a stairwell has an invalid ID or names an unknown plane",
            ),
            (
                // A step's stairwell is a stairwell of its plane, so the plane
                // exists too.
                format!(
                    "SELECT count(*) FROM stairwell_steps t WHERE NOT EXISTS (SELECT 1 FROM stairwells s WHERE s.plane = t.plane AND s.id = t.stairwell) \
                     OR typeof(t.layer) != 'integer' OR t.layer NOT BETWEEN -2147483648 AND 2147483647 \
                     OR typeof(t.x) != 'integer' OR t.x NOT BETWEEN -{MAX_SAFE_INTEGER} AND {MAX_SAFE_INTEGER} \
                     OR typeof(t.y) != 'integer' OR t.y NOT BETWEEN -{MAX_SAFE_INTEGER} AND {MAX_SAFE_INTEGER} \
                     OR {} IS NULL \
                     OR {} OR NOT EXISTS (SELECT 1 FROM definitions d WHERE d.type = 'proto' AND d.id = t.proto)",
                    step_index("t.step"),
                    not_identifier("t.proto")
                ),
                ErrorCode::StairwellStep,
                "a stairwell step names an unknown stairwell or proto, or has an invalid layer, position or index",
            ),
            (
                // Chunk `c` covers tiles `c * size ..= c * size + size - 1`.
                format!(
                    "SELECT count(*) FROM chunks c LEFT JOIN planes p ON p.id = c.plane WHERE p.id IS NULL \
                     OR typeof(c.layer) != 'integer' OR c.layer NOT BETWEEN -2147483648 AND 2147483647 \
                     OR typeof(c.cx) != 'integer' OR typeof(c.cy) != 'integer' \
                     OR c.cx NOT BETWEEN -({MAX_SAFE_INTEGER} / p.chunk_size) AND ({MAX_SAFE_INTEGER} - p.chunk_size + 1) / p.chunk_size \
                     OR c.cy NOT BETWEEN -({MAX_SAFE_INTEGER} / p.chunk_size) AND ({MAX_SAFE_INTEGER} - p.chunk_size + 1) / p.chunk_size"
                ),
                ErrorCode::Chunk,
                "a chunk names an unknown plane or has an invalid layer or position",
            ),
            (
                format!(
                    "SELECT count(*) FROM definitions WHERE {} OR {}",
                    not_identifier("type"),
                    not_identifier("id")
                ),
                ErrorCode::Definition,
                "a definition type or ID is not an identifier",
            ),
            (
                // The row's chunk exists, so its plane, layer and position
                // are valid too.
                format!(
                    "SELECT count(*) FROM object_uids u WHERE {} \
                     OR typeof(u.layer) != 'integer' OR typeof(u.cx) != 'integer' OR typeof(u.cy) != 'integer' \
                     OR typeof(u.object) != 'integer' OR u.object NOT BETWEEN 0 AND {MAXIMUM_OBJECT_INDEX} \
                     OR NOT EXISTS (SELECT 1 FROM chunks c WHERE c.plane = u.plane AND c.layer = u.layer AND c.cx = u.cx AND c.cy = u.cy)",
                    not_identifier("u.uid")
                ),
                ErrorCode::ObjectUid,
                "an object_uids row has an invalid uid or index, or names a missing chunk",
            ),
        ];
        for (query, code, problem) in checks {
            let count: i64 = self.connection.query_row(&query, [], |row| row.get(0))?;
            if count != 0 {
                return Err(Error::Rejected(code, problem.into()));
            }
        }
        Ok(())
    }

    /// Checks that each plane has at most as many zones as one source
    /// document holds, and that their parent chains end at a root. Planes
    /// are checked one at a time, so what SQLite keeps while following the
    /// chains is bounded by that limit rather than by the pack.
    /// [`Pack::check_rows`] has already checked that every zone's plane
    /// and parent exist.
    fn check_zone_trees(&self) -> Result<(), Error> {
        let limit = i64::try_from(Limits::default().max_array_items).unwrap_or(i64::MAX);
        let mut planes = self
            .connection
            .prepare("SELECT id FROM planes ORDER BY id")?;
        let mut count = self
            .connection
            .prepare("SELECT count(*) FROM zones WHERE plane = ?1")?;
        // The plane's zones reached from a root by following children: with
        // one parent each, a zone is reached once or, on or below a cycle,
        // never.
        let mut unrooted = self.connection.prepare(
            "WITH RECURSIVE children(parent, id) AS MATERIALIZED \
             (SELECT parent, id FROM zones WHERE plane = ?1 AND parent IS NOT NULL), \
             rooted(id) AS (SELECT id FROM zones WHERE plane = ?1 AND parent IS NULL \
             UNION ALL SELECT c.id FROM rooted r JOIN children c ON c.parent = r.id) \
             SELECT (SELECT count(*) FROM zones WHERE plane = ?1) - (SELECT count(*) FROM rooted)",
        )?;
        let mut rows = planes.query([])?;
        while let Some(row) = rows.next()? {
            let plane: String = row.get(0)?;
            let zones: i64 = count.query_row([&plane], |row| row.get(0))?;
            if zones > limit {
                return Err(Error::Rejected(
                    ErrorCode::Zone,
                    format!("plane {plane} has more zones than a source document holds"),
                ));
            }
            let missing: i64 = unrooted.query_row([&plane], |row| row.get(0))?;
            if missing != 0 {
                return Err(Error::Rejected(
                    ErrorCode::Zone,
                    "zone parents form a cycle".into(),
                ));
            }
        }
        Ok(())
    }

    /// Checks that every string is UTF-8, one bounded row at a time; SQLite
    /// accepts any bytes as text.
    fn check_string_encoding(&self) -> Result<(), Error> {
        let mut statement = self
            .connection
            .prepare("SELECT value FROM strings ORDER BY id")?;
        let mut rows = statement.query([])?;
        while let Some(row) = rows.next()? {
            match row.get_ref(0)? {
                types::ValueRef::Text(bytes) if std::str::from_utf8(bytes).is_ok() => {}
                _ => {
                    return Err(Error::Rejected(
                        ErrorCode::String,
                        "a string is not UTF-8 text".into(),
                    ));
                }
            }
        }
        Ok(())
    }

    /// Checks that every `document` column holds canonical JSON of an
    /// object whose identity keys agree with its row.
    fn check_documents(&self) -> Result<(), Error> {
        let limit = maximum_document_bytes();
        let document = bounded_text("document");
        let mut planes = self.connection.prepare(&format!(
            "SELECT id, kind, chunk_size, {document} FROM planes ORDER BY id"
        ))?;
        let mut has_zone = self
            .connection
            .prepare("SELECT EXISTS (SELECT 1 FROM zones WHERE plane = ?1 AND id = ?2)")?;
        let mut rows = planes.query([limit])?;
        while let Some(row) = rows.next()? {
            let (id, kind, chunk_size): (String, String, i64) =
                (row.get(0)?, row.get(1)?, row.get(2)?);
            let stored = stored_document(
                utf8_column(row, 3, ErrorCode::Plane)?,
                "a plane document",
                ErrorCode::Plane,
            )?;
            let default_zone_ok = match stored.root().get("default_zone") {
                None => true,
                Some(zone) => match zone.as_str() {
                    Some(zone) => has_zone.query_row(params![id, zone], |row| row.get(0))?,
                    None => false,
                },
            };
            if text(&stored, "format") != Some("tilewright.plane/1")
                || text(&stored, "id") != Some(&id)
                || text(&stored, "kind") != Some(&kind)
                || stored.root().get("chunk_size").and_then(Value::as_i64) != Some(chunk_size)
                || !default_zone_ok
            {
                return Err(Error::Rejected(
                    ErrorCode::Plane,
                    format!("the document of plane {id} disagrees with its row"),
                ));
            }
        }
        let mut zones = self.connection.prepare(&format!(
            "SELECT plane, id, parent, {document} FROM zones ORDER BY plane, id"
        ))?;
        let mut rows = zones.query([limit])?;
        while let Some(row) = rows.next()? {
            let (plane, id, parent): (String, String, Option<String>) =
                (row.get(0)?, row.get(1)?, row.get(2)?);
            let stored = stored_document(
                utf8_column(row, 3, ErrorCode::Zone)?,
                "a zone document",
                ErrorCode::Zone,
            )?;
            if text(&stored, "id") != Some(&id)
                || stored.root().get("parent").map(Value::as_str) != parent.as_deref().map(Some)
            {
                return Err(Error::Rejected(
                    ErrorCode::Zone,
                    format!("the document of zone {plane}/{id} disagrees with its row"),
                ));
            }
        }
        let mut definitions = self.connection.prepare(&format!(
            "SELECT type, id, {document} FROM definitions ORDER BY type, id"
        ))?;
        let mut rows = definitions.query([limit])?;
        while let Some(row) = rows.next()? {
            let (kind, id): (String, String) = (row.get(0)?, row.get(1)?);
            let stored = stored_document(
                utf8_column(row, 2, ErrorCode::Definition)?,
                "a definition document",
                ErrorCode::Definition,
            )?;
            if text(&stored, "format") != Some("tilewright.def/1")
                || text(&stored, "type") != Some(&kind)
                || text(&stored, "id") != Some(&id)
            {
                return Err(Error::Rejected(
                    ErrorCode::Definition,
                    format!("the document of definition {kind}/{id} disagrees with its row"),
                ));
            }
        }
        Ok(())
    }

    /// Checks every stored stairwell document against its row and the
    /// `stairwell_steps` table: each step is the row at its tile, naming its
    /// stairwell, its index and its prototype, and the table holds no other
    /// rows. One document is held at a time, and each step is one lookup.
    /// [`Pack::check_rows`] has already checked the rows' columns.
    fn check_stairwells(&self) -> Result<(), Error> {
        let mut stairwells = self.connection.prepare(&format!(
            "SELECT plane, id, {} FROM stairwells ORDER BY plane, id",
            bounded_text("document")
        ))?;
        let mut step_at = self.connection.prepare(
            "SELECT stairwell, step, proto FROM stairwell_steps WHERE plane = ?1 AND layer = ?2 AND x = ?3 AND y = ?4",
        )?;
        let mut listed: u64 = 0;
        // Stairwells and steps of the plane being checked: one source
        // document holds at most the array limit of stairwells and the value
        // limit of steps.
        let limits = Limits::default();
        let (mut current, mut count, mut steps) = (String::new(), 0_usize, 0_usize);
        let mut rows = stairwells.query([maximum_document_bytes()])?;
        while let Some(row) = rows.next()? {
            let (plane, id): (String, String) = (row.get(0)?, row.get(1)?);
            if plane != current {
                current.clone_from(&plane);
                (count, steps) = (0, 0);
            }
            count += 1;
            if count > limits.max_array_items {
                return Err(Error::Rejected(
                    ErrorCode::Stairwell,
                    format!("plane {plane} has more stairwells than a source document holds"),
                ));
            }
            let stored = stored_stairwell(&stored_document(
                utf8_column(row, 2, ErrorCode::Stairwell)?,
                "a stairwell document",
                ErrorCode::Stairwell,
            )?)?;
            steps = steps.saturating_add(stored.steps.len());
            if steps > limits.max_values {
                return Err(Error::Rejected(
                    ErrorCode::Stairwell,
                    format!("plane {plane} has more stairwell steps than a source document holds"),
                ));
            }
            let disagrees = || {
                Error::Rejected(
                    ErrorCode::Stairwell,
                    format!("the document of stairwell {plane}/{id} disagrees with its rows"),
                )
            };
            if stored.id != id {
                return Err(disagrees());
            }
            for (index, step) in stored.steps.iter().enumerate() {
                let found: Option<(String, i64, String)> = step_at
                    .query_row(params![plane, step.layer, step.at[0], step.at[1]], |row| {
                        Ok((row.get(0)?, row.get(1)?, row.get(2)?))
                    })
                    .optional()?;
                let expected = i64::try_from(index).map_err(|_| disagrees())?;
                if found.as_ref().is_none_or(|(stairwell, step_index, proto)| {
                    *stairwell != id || *step_index != expected || *proto != step.proto
                }) {
                    return Err(disagrees());
                }
            }
            listed = listed.saturating_add(u64::try_from(stored.steps.len()).unwrap_or(u64::MAX));
        }
        let rows: i64 =
            self.connection
                .query_row("SELECT count(*) FROM stairwell_steps", [], |row| row.get(0))?;
        if u64::try_from(rows).ok() != Some(listed) {
            return Err(Error::Rejected(
                ErrorCode::StairwellStep,
                "stairwell_steps holds a step that no stairwell lists".into(),
            ));
        }
        Ok(())
    }

    /// Render and ground layer counts from the stored profile, which must be
    /// canonical and agree with `profile_id` and `profile_version`.
    fn profile_layers(&self) -> Result<(usize, usize, Option<u32>), Error> {
        let profile = stored_document(
            self.meta("profile")?,
            "the stored profile",
            ErrorCode::Profile,
        )?;
        // The identity must be one a source profile can have: an identifier
        // and a version of at most 64 characters.
        let identity_ok = text(&profile, "id").is_some_and(is_identifier)
            && text(&profile, "version").is_some_and(|version| version.chars().count() <= 64);
        if text(&profile, "format") != Some("tilewright.profile/1")
            || !identity_ok
            || text(&profile, "id") != self.meta("profile_id")?.as_deref()
            || text(&profile, "version") != self.meta("profile_version")?.as_deref()
        {
            return Err(Error::Rejected(
                ErrorCode::Profile,
                "the stored profile disagrees with profile_id or profile_version".into(),
            ));
        }
        if !projection_ok(&profile) {
            return Err(Error::Rejected(
                ErrorCode::Profile,
                "the stored profile's projection is not one a source profile can have".into(),
            ));
        }
        let layer_height = stored_layer_height(&profile)?;
        // Each list holds unique identifiers within its count, and every
        // ground layer is a render layer other than `height` or `zone`.
        let layers = |key: &str, range: std::ops::RangeInclusive<usize>| {
            let names: Option<Vec<&str>> = profile
                .root()
                .get(key)
                .and_then(Value::as_array)
                .filter(|items| range.contains(&items.len()))
                .and_then(|items| items.iter().map(Value::as_str).collect());
            names
                .filter(|names| {
                    let unique: BTreeSet<&str> = names.iter().copied().collect();
                    unique.len() == names.len() && names.iter().all(|name| is_identifier(name))
                })
                .ok_or_else(|| {
                    Error::Rejected(
                        ErrorCode::Profile,
                        format!(
                            "stored profile does not list {}..={} unique {key}",
                            range.start(),
                            range.end()
                        ),
                    )
                })
        };
        let render = layers("render_layers", 1..=usize::from(u16::MAX - 1))?;
        let ground = layers("ground_layers", 0..=MAXIMUM_GROUND_LAYERS)?;
        let known: BTreeSet<&str> = render.iter().copied().collect();
        if ground
            .iter()
            .any(|name| !known.contains(name) || ["height", "zone"].contains(name))
        {
            return Err(Error::Rejected(
                ErrorCode::Profile,
                "a stored ground layer is not a render layer other than height or zone".into(),
            ));
        }
        Ok((render.len(), ground.len(), layer_height))
    }
}

/// A pack image being read into SQLite's copy. Reading it never fails:
/// once the image fails or ends, the rest of the copy is zeros, and whether
/// it ended early or the error it failed with is kept for the caller.
/// `rusqlite` frees its copy only once the copy is read in full.
struct Filled<R> {
    inner: R,
    missing: bool,
    failure: Option<std::io::Error>,
    /// The first [`HEADER_BYTES`] bytes `inner` yielded.
    head: Vec<u8>,
}

impl<R: std::io::Read> std::io::Read for Filled<R> {
    fn read(&mut self, buffer: &mut [u8]) -> std::io::Result<usize> {
        if buffer.is_empty() {
            return Ok(0);
        }
        while !self.missing && self.failure.is_none() {
            match self.inner.read(buffer) {
                Ok(0) => self.missing = true,
                Ok(read) => {
                    let wanted = HEADER_BYTES.saturating_sub(self.head.len()).min(read);
                    self.head.extend_from_slice(&buffer[..wanted]);
                    return Ok(read);
                }
                Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
                Err(error) => self.failure = Some(error),
            }
        }
        buffer.fill(0);
        Ok(buffer.len())
    }
}

/// How much of a pack's start [`check_header`] reads.
const HEADER_BYTES: usize = 20;

/// Checks the start of a pack, `head`, before SQLite reads it. A database
/// file begins with SQLite's header string, and its header bytes 18 and 19,
/// the file format write and read versions, are 1 for a rollback journal,
/// as in every pack, or 2 for WAL. SQLite opens a WAL database from an
/// immutable file but not from an in-memory copy, so every reader refuses
/// one, as from any other version, before SQLite reads it. Bytes that are
/// not a database are left for SQLite to refuse.
fn check_header(head: &[u8]) -> Result<(), Error> {
    if head.len() == HEADER_BYTES
        && head.starts_with(b"SQLite format 3\0")
        && (head[18] != 1 || head[19] != 1)
    {
        return Err(Error::Rejected(
            ErrorCode::Identity,
            "the SQLite header's file format versions are not 1, a rollback journal's".into(),
        ));
    }
    Ok(())
}

/// Whether `read` yields at least one more byte.
fn read_some(read: &mut impl std::io::Read) -> Result<bool, Error> {
    loop {
        match read.read(&mut [0]) {
            Ok(read) => return Ok(read != 0),
            Err(error) if error.kind() == std::io::ErrorKind::Interrupted => {}
            Err(error) => return Err(Error::Io(error)),
        }
    }
}

/// Puts a connection to a pack in SQLite's defensive mode, with untrusted
/// schemas disabled and writes refused.
fn harden(connection: &Connection) -> Result<(), Error> {
    connection.set_db_config(DbConfig::SQLITE_DBCONFIG_DEFENSIVE, true)?;
    connection.set_db_config(DbConfig::SQLITE_DBCONFIG_TRUSTED_SCHEMA, false)?;
    connection.execute_batch("PRAGMA query_only = ON")?;
    Ok(())
}

/// Checks that a pack of `length` bytes is exactly its pages, and that they
/// are within the pack size limit. Nothing, not even a sparse hole, follows
/// the last page, so hashing a pack never reads past its pages.
fn check_pages(connection: &Connection, length: u64, limits: &VerifyLimits) -> Result<(), Error> {
    let pages: i64 = connection.query_row("PRAGMA page_count", [], |row| row.get(0))?;
    if u64::try_from(pages)
        .ok()
        .and_then(|pages| pages.checked_mul(PAGE_SIZE.unsigned_abs()))
        != Some(length)
    {
        return Err(Error::Rejected(
            ErrorCode::FileLength,
            format!("the file is {length} bytes, not its {pages} pages of {PAGE_SIZE} bytes"),
        ));
    }
    if length > limits.maximum_pack_bytes {
        return Err(too_long(limits.maximum_pack_bytes));
    }
    Ok(())
}

/// Checks the schema, application ID, user version and `meta` of a pack
/// before any other query runs.
fn check_identity(connection: &Connection) -> Result<(), Error> {
    // At most one row more than expected, each value bounded, is loaded.
    let mut schema = connection.prepare(&format!(
        "SELECT {}, {}, {} FROM sqlite_schema WHERE sql IS NOT NULL ORDER BY name LIMIT {}",
        bounded_text("type"),
        bounded_text("name"),
        bounded_text("sql"),
        SCHEMA_TABLES.len() + 1
    ))?;
    let mut found: Vec<(Option<String>, Option<String>, Option<String>)> = Vec::new();
    let mut rows = schema.query([bound(MAXIMUM_SCHEMA_BYTES)])?;
    while let Some(row) = rows.next()? {
        found.push((
            utf8_column(row, 0, ErrorCode::Schema)?,
            utf8_column(row, 1, ErrorCode::Schema)?,
            utf8_column(row, 2, ErrorCode::Schema)?,
        ));
    }
    drop(rows);
    drop(schema);
    let mut expected: Vec<(Option<String>, Option<String>, Option<String>)> = SCHEMA_TABLES
        .iter()
        .map(|(name, sql)| {
            (
                Some("table".to_owned()),
                Some((*name).to_owned()),
                Some((*sql).to_owned()),
            )
        })
        .collect();
    expected.sort_by(|left, right| left.1.cmp(&right.1));
    if found != expected {
        return Err(Error::Rejected(
            ErrorCode::Schema,
            "the pack schema does not match this format".into(),
        ));
    }
    let application: i32 = connection.query_row("PRAGMA application_id", [], |row| row.get(0))?;
    let version: i32 = connection.query_row("PRAGMA user_version", [], |row| row.get(0))?;
    let page_size: i64 = connection.query_row("PRAGMA page_size", [], |row| row.get(0))?;
    if application != APPLICATION_ID || version != USER_VERSION || page_size != PAGE_SIZE {
        return Err(Error::Rejected(
            ErrorCode::Identity,
            format!(
                "not a Tilewright pack of user_version {USER_VERSION} with {PAGE_SIZE}-byte pages"
            ),
        ));
    }
    // `VACUUM` leaves a compiled pack no free pages. Free pages hold no
    // data, so a sparse file could make them holes that cost nothing to
    // store and much to hash; with none, the integrity check reads every
    // page as table data before the file is hashed.
    let free: i64 = connection.query_row("PRAGMA freelist_count", [], |row| row.get(0))?;
    if free != 0 {
        return Err(Error::Rejected(
            ErrorCode::FreePages,
            format!("the pack has {free} free pages; a pack has none"),
        ));
    }
    let meta = read_meta(connection)?;
    let value = |key: &str| {
        meta.iter()
            .find(|(name, _)| name == key)
            .map(|(_, value)| value.as_str())
    };
    let digest_ok = value("source_digest").is_some_and(|digest| {
        digest.len() == 64
            && digest
                .bytes()
                .all(|byte| byte.is_ascii_digit() || (b'a'..=b'f').contains(&byte))
    });
    if value("format") != Some(FORMAT)
        || value("chunk_encoding") != Some(CHUNK_ENCODING)
        || !digest_ok
    {
        return Err(Error::Rejected(
            ErrorCode::Meta,
            format!("the pack's metadata is not that of {FORMAT}"),
        ));
    }
    Ok(())
}

/// Reads `meta` after checking, without loading any value, that it holds
/// exactly the format's keys with text values of bounded size.
fn read_meta(connection: &Connection) -> Result<Vec<(String, String)>, Error> {
    let keys = META_KEYS
        .iter()
        .map(|key| format!("'{key}'"))
        .collect::<Vec<_>>()
        .join(", ");
    let (total, known, oversized): (i64, i64, i64) = connection.query_row(
        &format!(
            "SELECT count(*), count(*) FILTER (WHERE key IN ({keys})), \
             count(*) FILTER (WHERE typeof(value) != 'text' OR octet_length(value) > CASE key WHEN 'profile' THEN ?1 ELSE ?2 END) \
             FROM meta"
        ),
        params![maximum_document_bytes(), bound(MAXIMUM_META_BYTES)],
        |row| Ok((row.get(0)?, row.get(1)?, row.get(2)?)),
    )?;
    let expected = i64::try_from(META_KEYS.len()).unwrap_or(i64::MAX);
    if total != expected || known != expected || oversized != 0 {
        return Err(Error::Rejected(
            ErrorCode::Meta,
            format!("the pack's metadata is not that of {FORMAT}"),
        ));
    }
    let mut statement = connection.prepare("SELECT key, value FROM meta ORDER BY key")?;
    let mut rows = statement.query([])?;
    let mut meta = Vec::with_capacity(META_KEYS.len());
    while let Some(row) = rows.next()? {
        // Every value was checked to be text above.
        let value = utf8_column(row, 1, ErrorCode::Meta)?.unwrap_or_default();
        meta.push((row.get(0)?, value));
    }
    Ok(meta)
}

/// What chunk records may reference. Strings are checked by ID, since
/// verification has proven IDs are `1..=strings`. Every other lookup is
/// cached per distinct ID in a cache of bounded size, so checking an object
/// costs map lookups and at worst a query per object: memory stays bounded
/// and work stays proportional to the decoded bytes its budget bounds.
struct Targets<'c> {
    strings: u32,
    render_layers: usize,
    /// Whether string `?1` names a plane.
    is_plane: Statement<'c>,
    /// Whether string `?1` names a `proto` definition.
    is_proto: Statement<'c>,
    /// Whether string `?1` names a zone of plane `?2`.
    is_zone: Statement<'c>,
    /// Whether plane `?1` (a string ID, or 0 for plane `?2`) has the anchor
    /// named by string `?3`.
    has_anchor: Statement<'c>,
    /// The value of string `?1`.
    value: Statement<'c>,
    planes: BTreeMap<u32, bool>,
    protos: BTreeMap<u32, bool>,
    field_names: BTreeMap<u32, bool>,
    /// Anchors by `(plane, anchor)` for portals that name their plane.
    anchors: BTreeMap<(u32, u32), bool>,
    /// The plane of the chunks being checked, and the lookups relative to
    /// it: its zones and its own anchors.
    current: String,
    zones: BTreeMap<u32, bool>,
    own_anchors: BTreeMap<u32, bool>,
    /// The `object_uids` row of uid `?1`.
    uid_row: Statement<'c>,
    /// The layer and position of the chunk being checked, the index of its
    /// next object in the record's depth-first order, and the objects with
    /// a uid checked so far.
    chunk: (i32, i64, i64),
    index: i64,
    uids: u64,
}

impl<'c> Targets<'c> {
    /// The lookups of a verification whose strings are `1..=strings` and
    /// whose profile has `render_layers` render layers.
    fn new(connection: &'c Connection, strings: u32, render_layers: usize) -> Result<Self, Error> {
        Ok(Self {
            strings,
            render_layers,
            is_plane: connection.prepare(
                "SELECT EXISTS (SELECT 1 FROM strings s JOIN planes p ON p.id = s.value WHERE s.id = ?1)",
            )?,
            is_proto: connection.prepare(
                "SELECT EXISTS (SELECT 1 FROM strings s JOIN definitions d ON d.type = 'proto' AND d.id = s.value WHERE s.id = ?1)",
            )?,
            is_zone: connection.prepare(
                "SELECT EXISTS (SELECT 1 FROM strings s JOIN zones z ON z.plane = ?2 AND z.id = s.value WHERE s.id = ?1)",
            )?,
            has_anchor: connection.prepare(
                "SELECT EXISTS (SELECT 1 FROM strings n JOIN anchors a ON a.id = n.value \
                 WHERE n.id = ?3 AND a.plane = CASE WHEN ?1 = 0 THEN ?2 ELSE (SELECT value FROM strings WHERE id = ?1) END)",
            )?,
            value: connection.prepare("SELECT value FROM strings WHERE id = ?1")?,
            uid_row: connection.prepare(
                "SELECT plane, layer, cx, cy, object FROM object_uids WHERE uid = ?1",
            )?,
            chunk: (0, 0, 0),
            index: 0,
            uids: 0,
            planes: BTreeMap::new(),
            protos: BTreeMap::new(),
            field_names: BTreeMap::new(),
            anchors: BTreeMap::new(),
            current: String::new(),
            zones: BTreeMap::new(),
            own_anchors: BTreeMap::new(),
        })
    }
}

/// Entries kept per lookup cache. A full cache is cleared, so memory stays
/// bounded however many distinct targets a pack names, and a lookup is at
/// worst one query per object.
const CACHE_ENTRIES: usize = 1 << 16;

/// A cached boolean lookup.
fn cached<K: Ord + Copy>(
    cache: &mut BTreeMap<K, bool>,
    key: K,
    lookup: impl FnOnce() -> Result<bool, Error>,
) -> Result<bool, Error> {
    if let Some(known) = cache.get(&key) {
        return Ok(*known);
    }
    let found = lookup()?;
    if cache.len() >= CACHE_ENTRIES {
        cache.clear();
    }
    cache.insert(key, found);
    Ok(found)
}

impl Targets<'_> {
    fn string(&self, id: u32) -> bool {
        (1..=self.strings).contains(&id)
    }

    /// Starts checking the chunk of `plane` at `chunk`, dropping lookups
    /// relative to the previous plane.
    fn enter(&mut self, plane: &str, chunk: (i32, i64, i64)) {
        if self.current != plane {
            plane.clone_into(&mut self.current);
            self.zones.clear();
            self.own_anchors.clear();
        }
        self.chunk = chunk;
        self.index = 0;
    }

    /// Checks that an object's uid has the `object_uids` row that names the
    /// object: the current chunk and the object's index in it.
    fn uid(&mut self, object: &ObjectRecord) -> Result<(), Error> {
        let index = self.index;
        self.index += 1;
        let Some(uid) = &object.uid else {
            return Ok(());
        };
        self.uids += 1;
        let row: Option<(
            types::Value,
            types::Value,
            types::Value,
            types::Value,
            types::Value,
        )> = self
            .uid_row
            .query_row([uid], |row| {
                Ok((
                    row.get(0)?,
                    row.get(1)?,
                    row.get(2)?,
                    row.get(3)?,
                    row.get(4)?,
                ))
            })
            .optional()?;
        let (layer, cx, cy) = self.chunk;
        let expected = (
            types::Value::Text(self.current.clone()),
            types::Value::Integer(i64::from(layer)),
            types::Value::Integer(cx),
            types::Value::Integer(cy),
            types::Value::Integer(index),
        );
        if row.as_ref() == Some(&expected) {
            Ok(())
        } else {
            Err(Error::Rejected(
                ErrorCode::ObjectUid,
                format!("the object_uids row of {uid} does not name its object"),
            ))
        }
    }

    fn proto(&mut self, id: u32) -> Result<bool, Error> {
        if !self.string(id) {
            return Ok(false);
        }
        let statement = &mut self.is_proto;
        cached(&mut self.protos, id, || {
            Ok(statement.query_row([id], |row| row.get(0))?)
        })
    }

    fn field_name(&mut self, id: u32) -> Result<bool, Error> {
        if !self.string(id) {
            return Ok(false);
        }
        let statement = &mut self.value;
        cached(&mut self.field_names, id, || {
            let name: String = statement.query_row([id], |row| row.get(0))?;
            Ok(is_field_name(&name))
        })
    }

    /// Checks that every palette entry is used, that ground cells name
    /// `proto` definitions, and that zone cells name zones of the current
    /// plane.
    fn check_palette(&mut self, chunk: &ChunkData) -> Result<(), Error> {
        fn mark(cells: &[u16], used: &mut [bool]) {
            for cell in cells {
                if let Some(flag) = usize::from(*cell)
                    .checked_sub(1)
                    .and_then(|index| used.get_mut(index))
                {
                    *flag = true;
                }
            }
        }
        let mut ground = vec![false; chunk.palette.len()];
        let mut zone = vec![false; chunk.palette.len()];
        for grid in &chunk.ground {
            mark(grid, &mut ground);
        }
        mark(&chunk.zones, &mut zone);
        for ((id, ground), zone) in chunk.palette.iter().zip(ground).zip(zone) {
            let id = *id;
            let mut ok = (ground || zone) && self.string(id) && (!ground || self.proto(id)?);
            if ok && zone {
                let (statement, plane) = (&mut self.is_zone, &self.current);
                ok = cached(&mut self.zones, id, || {
                    Ok(statement.query_row(params![id, plane], |row| row.get(0))?)
                })?;
            }
            if !ok {
                return Err(Error::Rejected(
                    ErrorCode::Reference,
                    "a palette entry is unused or not a proto definition or zone of its plane"
                        .into(),
                ));
            }
        }
        Ok(())
    }

    fn portal(&mut self, portal: Option<&PortalRecord>) -> Result<bool, Error> {
        Ok(match portal {
            Some(PortalRecord::Location { x, y, .. })
                if [x, y]
                    .iter()
                    .any(|value| !(-MAX_SAFE_INTEGER..=MAX_SAFE_INTEGER).contains(*value)) =>
            {
                false
            }
            Some(PortalRecord::Location { plane: 0, .. }) | None => true,
            Some(PortalRecord::Location { plane, .. }) => {
                let plane = *plane;
                if !self.string(plane) {
                    return Ok(false);
                }
                let statement = &mut self.is_plane;
                cached(&mut self.planes, plane, || {
                    Ok(statement.query_row([plane], |row| row.get(0))?)
                })?
            }
            Some(PortalRecord::Anchor { plane, anchor }) => {
                let (plane, anchor) = (*plane, *anchor);
                if !(plane == 0 || self.string(plane)) || !self.string(anchor) {
                    return Ok(false);
                }
                let (statement, current) = (&mut self.has_anchor, &self.current);
                let lookup =
                    || Ok(statement.query_row(params![plane, current, anchor], |row| row.get(0))?);
                if plane == 0 {
                    cached(&mut self.own_anchors, anchor, lookup)?
                } else {
                    cached(&mut self.anchors, (plane, anchor), lookup)?
                }
            }
        })
    }

    fn check(&mut self, object: &ObjectRecord) -> Result<(), Error> {
        self.uid(object)?;
        let mut ok = self.proto(object.proto)?
            && usize::from(object.layer) <= self.render_layers
            && self.portal(object.portal.as_ref())?;
        // Field names are field-name strings, each at most once per object.
        let mut names = BTreeSet::new();
        for (name, value) in &object.fields {
            ok &= names.insert(*name) && self.field_name(*name)?;
            match value {
                FieldRecord::String(id) => ok &= self.string(*id),
                FieldRecord::Integer(number) => {
                    ok &= (-MAX_SAFE_INTEGER..=MAX_SAFE_INTEGER).contains(number);
                }
                _ => {}
            }
        }
        if ok {
            Ok(())
        } else {
            Err(Error::Rejected(
                ErrorCode::Reference,
                "an object references a missing string, layer, plane or anchor, repeats or misnames a field, or has a portal position or integer field out of range".into(),
            ))
        }
    }
}

#[cfg(test)]
mod tests {
    use super::{CACHE_ENTRIES, Pack, cached};
    use sha2::Digest as _;
    use std::{collections::BTreeMap, path::Path};

    #[test]
    fn lookup_caches_stay_bounded() {
        let mut cache = BTreeMap::new();
        for key in 0..u32::try_from(CACHE_ENTRIES * 3).unwrap() {
            assert!(cached(&mut cache, key, || Ok(true)).unwrap());
            assert!(cache.len() <= CACHE_ENTRIES);
        }
        // The latest keys are kept and not looked up again; evicted ones are.
        let last = u32::try_from(CACHE_ENTRIES * 3 - 1).unwrap();
        let mut again = false;
        let kept = cached(&mut cache, last, || {
            again = true;
            Ok(false)
        });
        assert!(kept.unwrap() && !again);
        let mut queried = false;
        let evicted = cached(&mut cache, 0, || {
            queried = true;
            Ok(false)
        });
        assert!(!evicted.unwrap() && queried);
    }

    #[test]
    fn hashing_an_image_reads_sqlite_copy_in_place() {
        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join(
            "../../fixtures/packs/5830e26e449a1f0d0c786b32801b7ed68e351777db5f586532478cb4384844b2.twpack",
        );
        let bytes = std::fs::read(path).unwrap();
        let pack = Pack::open_bytes(&bytes, None, &super::VerifyLimits::default()).unwrap();
        // The digest reads the buffer SQLite reads, without allocating a
        // second image: serializing it is a view of that buffer.
        let image = pack.connection.serialize(rusqlite::MAIN_DB).unwrap();
        assert!(matches!(image, rusqlite::serialize::Data::Shared(_)));
        assert_eq!(&*image, bytes.as_slice());
    }

    #[test]
    fn images_are_read_only() {
        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join(
            "../../fixtures/packs/5830e26e449a1f0d0c786b32801b7ed68e351777db5f586532478cb4384844b2.twpack",
        );
        let bytes = std::fs::read(path).unwrap();
        let pack = Pack::open_bytes(&bytes, None, &super::VerifyLimits::default()).unwrap();
        // Even with query_only turned off, the image cannot be written.
        pack.connection
            .execute_batch("PRAGMA query_only = OFF")
            .unwrap();
        for write in [
            "DELETE FROM strings",
            "CREATE TABLE extra (value)",
            "PRAGMA user_version = 3",
        ] {
            assert!(pack.connection.execute_batch(write).is_err(), "{write}");
        }
        pack.verify().unwrap();
        assert_eq!(
            pack.file_digest().unwrap(),
            <[u8; 32]>::from(sha2::Sha256::digest(&bytes))
        );
    }

    /// Objects at any depth and their fields.
    fn count(objects: &[crate::ObjectRecord]) -> u64 {
        objects
            .iter()
            .map(|object| {
                1 + u64::try_from(object.fields.len()).unwrap() + count(&object.inventory)
            })
            .sum()
    }

    /// Objects at any depth.
    fn count_objects(objects: &[crate::ObjectRecord]) -> u64 {
        objects
            .iter()
            .map(|object| 1 + count_objects(&object.inventory))
            .sum()
    }

    #[test]
    fn chunk_reads_keep_to_their_entry_budget() {
        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join(
            "../../fixtures/packs/5830e26e449a1f0d0c786b32801b7ed68e351777db5f586532478cb4384844b2.twpack",
        );
        let pack = Pack::open(&path).unwrap();
        let chunk = pack.chunk("overworld", 0, 0, 0).unwrap().unwrap();
        let entries = count(&chunk.objects);
        assert!(
            entries > count_objects(&chunk.objects),
            "the chunk has fields"
        );
        assert_eq!(
            pack.chunk_within("overworld", 0, 0, 0, entries).unwrap(),
            Some(chunk)
        );
        // Fields count as well as objects.
        let error = pack
            .chunk_within("overworld", 0, 0, 0, entries - 1)
            .unwrap_err();
        assert_eq!(error.code(), crate::ErrorCode::DecodeBudget);
        // A chunk without objects fits any budget, and a missing chunk is
        // none.
        assert!(
            pack.chunk_within("overworld", 1, 0, 0, 0)
                .unwrap()
                .is_some()
        );
        assert!(
            pack.chunk_within("overworld", 0, 9, 9, 0)
                .unwrap()
                .is_none()
        );
        // A read of an object decodes its chunk within the budget too.
        let uid = "o.k3vq7zr2mxa5d6wfh4pnyt2cgu";
        assert!(pack.object_within(uid, entries).unwrap().is_some());
        let error = pack.object_within(uid, entries - 1).unwrap_err();
        assert_eq!(error.code(), crate::ErrorCode::DecodeBudget);
    }

    #[test]
    fn stairwell_reads_keep_to_their_step_budget() {
        let path = Path::new(env!("CARGO_MANIFEST_DIR")).join(
            "../../fixtures/packs/5830e26e449a1f0d0c786b32801b7ed68e351777db5f586532478cb4384844b2.twpack",
        );
        let pack = Pack::open(&path).unwrap();
        let mut stairwells = Vec::new();
        pack.visit_stairwells("overworld", crate::Page::all(), &mut |stairwell| {
            stairwells.push(stairwell);
            Ok(())
        })
        .unwrap();
        let longest = stairwells
            .iter()
            .max_by_key(|stairwell| stairwell.steps.len())
            .unwrap();
        let steps = u64::try_from(longest.steps.len()).unwrap();
        assert_eq!(
            pack.stairwell_within("overworld", &longest.id, steps)
                .unwrap()
                .as_ref(),
            Some(longest)
        );
        let error = pack
            .stairwell_within("overworld", &longest.id, steps - 1)
            .unwrap_err();
        assert_eq!(error.code(), crate::ErrorCode::DecodeBudget);
        let mut listed = 0;
        pack.visit_stairwells_within("overworld", crate::Page::all(), steps, &mut |_| {
            listed += 1;
            Ok(())
        })
        .unwrap();
        assert_eq!(listed, stairwells.len());
        let error = pack
            .visit_stairwells_within("overworld", crate::Page::all(), steps - 1, &mut |_| Ok(()))
            .unwrap_err();
        assert_eq!(error.code(), crate::ErrorCode::DecodeBudget);
    }

    #[test]
    fn packs_keep_the_absolute_path_they_opened() {
        // Unit tests run in the package directory, so this path is relative.
        let target = Path::new(env!("CARGO_MANIFEST_DIR")).join("../../target");
        std::fs::create_dir_all(&target).unwrap();
        let directory = tempfile::tempdir_in(&target).unwrap();
        let world = tilewright_source::load(
            &Path::new(env!("CARGO_MANIFEST_DIR")).join("../../fixtures/basic"),
        )
        .unwrap();
        let written = crate::write_pack(&world, directory.path()).unwrap();
        let relative = Path::new("../../target")
            .join(directory.path().file_name().unwrap())
            .join(written.path.file_name().unwrap());
        let pack = Pack::open(&relative).unwrap();
        let super::Source::File(path) = &pack.source else {
            panic!("a pack opened from a path has a file source");
        };
        assert!(path.is_absolute(), "{}", path.display());
        pack.verify().unwrap();
    }
}