-- SPDX-FileCopyrightText: © 2026 Vladimir Zorin <vladimir@deviant.guru>
-- SPDX-License-Identifier: LicenseRef-OWL-1.0-or-later
-- Licensed under OWL v1.0+. See LICENSE.

--[==[
Single-file embedded key-value database with multiple keyspaces, typed values,
TTL/expiry, sorted sets, lists, vector similarity search, BM25 full-text
search, and optional encryption at rest. All operations are synchronous.
Concurrency: single writer, multiple readers (file-level locking, mmap
snapshots). Database files use a copy-on-write B+ tree with double-buffered
meta pages for atomic commits.
]==]

local core = require("mneme.core")
-- We need random_bytes from litls.core
local litls = require("litls.core")
local crypto = require("mneme.crypto")

-- Value type tags (match C defines)
local TYPE_BYTES = 1
local TYPE_INTEGER = 2
local TYPE_FLOAT = 3
local TYPE_ZSET = 4
local TYPE_LIST = 5
local TYPE_VECTOR = 6
local TYPE_FT_META = 7

-- Metric IDs for vsearch_scan (match C switch)
local METRIC_COSINE = 1
local METRIC_DOT = 2
local METRIC_L2 = 3

-- Limits (match C defines)
local MAX_KEY_LEN = 4000
local MAX_KS_NAME_LEN = 255
local PAGE_SIZE = 4096

-- ── Encryption helpers ───────────────────────────────────────────────

-- Associated data for value AEAD: the B-tree key plus the unencrypted
-- metadata stored alongside the value (type tag, TTL). Binding all three
-- prevents an offline attacker from swapping ciphertexts between keys,
-- re-tagging an encrypted value to reinterpret its plaintext (e.g. int64
-- bytes as a string), or resurrecting/expiring an entry by editing its
-- TTL -- any such edit fails authentication on read. Every code path that
-- changes the tag or TTL of an encrypted value must therefore re-encrypt.
local enc_aad = function(btree_key, type_tag, ttl)
	return btree_key .. string.char(type_tag) .. core.pack_int64(ttl or 0)
end

-- Encrypt a value for an encrypted keyspace.
-- subkey: DEK userdata (from dek:derive), btree_key: the B-tree key,
-- val_str: raw value bytes, type_tag/ttl: the metadata stored with it.
-- Returns: encrypted_val (the type tag and TTL are stored unchanged).
local enc_value = function(subkey, btree_key, val_str, type_tag, ttl)
	local nonce = litls.random_bytes(12)
	local ct_tag = subkey:encrypt(nonce, enc_aad(btree_key, type_tag, ttl), val_str)
	return nonce .. ct_tag
end

-- Decrypt a value from an encrypted keyspace. type_tag/ttl must be the
-- values stored alongside the ciphertext (they are part of the AAD).
-- Returns: decrypted_val or nil, err
local dec_value = function(subkey, btree_key, enc_val, type_tag, ttl)
	if #enc_val < 28 then -- 12 nonce + 16 tag minimum
		return nil, "encrypted value too short"
	end
	local nonce = enc_val:sub(1, 12)
	local ct_tag = enc_val:sub(13)
	return subkey:decrypt(nonce, enc_aad(btree_key, type_tag, ttl), ct_tag)
end

-- ── Value packing/unpacking ───────────────────────────────────────────

local pack_value = function(value, explicit_type)
	if explicit_type == TYPE_INTEGER then
		return core.pack_int64(value), TYPE_INTEGER
	elseif explicit_type == TYPE_FLOAT then
		return core.pack_double(value), TYPE_FLOAT
	elseif explicit_type == TYPE_BYTES then
		return tostring(value), TYPE_BYTES
	end

	-- Auto-detect from Lua type
	local t = type(value)
	if t == "string" then
		return value, TYPE_BYTES
	elseif t == "number" then
		if value ~= value then -- NaN
			return core.pack_double(value), TYPE_FLOAT
		elseif value == math.huge or value == -math.huge then
			return core.pack_double(value), TYPE_FLOAT
		elseif math.floor(value) == value and value >= -2 ^ 53 and value <= 2 ^ 53 then
			return core.pack_int64(value), TYPE_INTEGER
		else
			return core.pack_double(value), TYPE_FLOAT
		end
	else
		return nil, nil, "unsupported value type: " .. t
	end
end

local unpack_value = function(raw, type_tag)
	if type_tag == TYPE_BYTES then
		return raw
	elseif type_tag == TYPE_INTEGER then
		return core.unpack_int64(raw)
	elseif type_tag == TYPE_FLOAT then
		return core.unpack_double(raw)
	else
		return raw
	end
end

-- ── Keyspace directory helpers ────────────────────────────────────────

local ks_dir_get = function(txn, dir_root, name)
	local val = txn:btree_get(dir_root, name)
	if not val then
		return nil
	end
	return core.unpack_u32(val)
end

local ks_dir_set = function(txn, dir_root, name, ks_root)
	return txn:btree_put(dir_root, name, core.pack_u32(ks_root), TYPE_BYTES, 0)
end

local ks_dir_del = function(txn, dir_root, name)
	return txn:btree_del(dir_root, name)
end

-- ── Internal: write within an open write txn ──────────────────────────

-- Per-transaction cache of (directory root, keyspace roots): without it
-- every ks_read/ks_write re-descends the directory B-tree, which compounds
-- badly in multi-op transactions (zs:add, batches, list ranges). Weak keys
-- let entries die with their txn userdata. Only ks_read/ks_write/ks_delete
-- may touch keyspace roots on a cached txn; code that mutates the
-- directory directly (compact, encryption metadata) uses dedicated txns.
local txn_root_cache = setmetatable({}, { __mode = "k" })

local cached_roots = function(txn)
	local c = txn_root_cache[txn]
	if not c then
		c = { dir_root = txn:meta().keyspace_root_pgno, roots = {} }
		txn_root_cache[txn] = c
	end
	return c
end

-- Performs a single btree_put within a transaction, updating the keyspace
-- directory and meta as needed. Returns true or nil, err.
local ks_write = function(ks_name, txn, key, val_str, type_tag, ttl)
	local c = cached_roots(txn)

	local ks_root = c.roots[ks_name]
	if not ks_root then
		ks_root = ks_dir_get(txn, c.dir_root, ks_name)
	end
	if not ks_root then
		local pg, pg_err = txn:alloc_leaf()
		if not pg then
			return nil, pg_err or "allocation failed"
		end
		ks_root = pg
	end

	local new_ks_root, put_err = txn:btree_put(ks_root, key, val_str, type_tag, ttl)
	if not new_ks_root then
		return nil, put_err
	end

	local new_dir_root = ks_dir_set(txn, c.dir_root, ks_name, new_ks_root)
	if not new_dir_root then
		return nil, "directory update failed"
	end
	if new_dir_root ~= c.dir_root then
		txn:set_ks_root(new_dir_root)
		c.dir_root = new_dir_root
	end
	c.roots[ks_name] = new_ks_root
	return true
end

-- Performs a btree_del within a transaction.
local ks_delete = function(ks_name, txn, key)
	local c = cached_roots(txn)
	local ks_root = c.roots[ks_name]
	if not ks_root then
		ks_root = ks_dir_get(txn, c.dir_root, ks_name)
		if not ks_root then
			return nil, "not found"
		end
	end

	local new_ks_root = txn:btree_del(ks_root, key)
	if not new_ks_root then
		return nil, "delete failed"
	end

	local new_dir_root = ks_dir_set(txn, c.dir_root, ks_name, new_ks_root)
	if not new_dir_root then
		return nil, "directory update failed"
	end
	if new_dir_root ~= c.dir_root then
		txn:set_ks_root(new_dir_root)
		c.dir_root = new_dir_root
	end
	c.roots[ks_name] = new_ks_root
	return true
end

-- Reads a key within a transaction. Returns raw_val, type_tag, ttl or nil, err.
local ks_read = function(txn, ks_name, key)
	local c = cached_roots(txn)
	local ks_root = c.roots[ks_name]
	if not ks_root then
		ks_root = ks_dir_get(txn, c.dir_root, ks_name)
		if not ks_root then
			return nil, "not found"
		end
		c.roots[ks_name] = ks_root
	end
	return txn:btree_get(ks_root, key)
end

-- ── Validation helpers ────────────────────────────────────────────────

-- Validate a user-supplied keyspace name.
-- Rules: string, 1–MAX_KS_NAME_LEN bytes, first byte must not be NUL.
local check_keyspace_name = function(name)
	if type(name) ~= "string" or #name == 0 or #name > MAX_KS_NAME_LEN then
		return nil, "keyspace name must be 1-255 byte string"
	end
	if name:sub(1, 1) == "\x00" then
		return nil, "invalid keyspace name"
	end
	return true
end

-- Reserved key for vector dimension metadata. The leading NUL keeps it
-- out of the user keyspace: plain keys reject NUL bytes (below), so no
-- user key can collide with it -- the same convention the keyspace
-- directory uses for its internal entries.
local VDIM_KEY = "\x00vdim"

local check_key = function(self, key)
	if type(key) ~= "string" then
		return nil, "key must be a string"
	end
	if #key == 0 then
		return nil, "key must not be empty"
	end
	if #key > MAX_KEY_LEN then
		return nil, "key too long"
	end
	-- NUL is the internal composite-key separator and reserved-key prefix:
	-- a plain key containing it could alias list/zset metadata, element
	-- keys, or the vector dimension record.
	if key:find("\x00", 1, true) then
		return nil, "key must not contain null bytes"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	return db
end

local check_write_key = function(self, key)
	local db, err = check_key(self, key)
	if not db then
		return nil, err
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end
	return db
end

local calc_ttl = function(opts)
	if opts and opts.ttl ~= nil then
		if type(opts.ttl) ~= "number" or opts.ttl < 0 then
			return nil, "ttl must be a non-negative number"
		end
		return os.time() + opts.ttl
	end
	return 0
end

local with_write_txn = function(db, fn)
	local txn, err = db.__state.handle:txn_begin(false)
	if not txn then
		return nil, err
	end

	-- A raised error must not leak the txn: it holds the write flock
	-- and blocks every future write txn on this handle until GC.
	local pok, ok, r1, r2 = pcall(fn, txn)
	if not pok then
		txn:abort()
		return nil, ok
	end
	if ok == nil then
		txn:abort()
		return nil, r1
	end
	if ok == false then
		txn:abort()
		return r1, r2
	end

	local cok, cerr = txn:commit()
	if not cok then
		return nil, cerr
	end
	return r1, r2
end

-- ── Keyspace methods ──────────────────────────────────────────────────

---! Get the value for a key
---@ ks:get(`key`{.str}) -> `value`{.str .or_nil}, `err`{.str .err}
--[===[
Returns the value associated with the key, or nil if not found or expired.
Type is preserved: integers return as numbers, floats as numbers, bytes as
strings. Encrypted keyspaces decrypt transparently.
]===]
local ks_get = function(self, key)
	local db, cerr = check_key(self, key)
	if not db then
		return nil, cerr
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local raw, tag, ttl = ks_read(txn, self.cfg.name, key)
	txn:abort()

	if not raw then
		return nil, "not found"
	end

	-- TTL check: if expired, return nil (lazy detection, no delete)
	if ttl and ttl ~= 0 and ttl < os.time() then
		return nil, "not found"
	end

	-- Decrypt if encrypted keyspace
	if self.__state.subkey then
		local dec, derr = dec_value(self.__state.subkey, key, raw, tag, ttl)
		if not dec then
			return nil, derr
		end
		raw = dec
	end

	return unpack_value(raw, tag)
end

---! Set a key-value pair with optional TTL
---@ ks:set(`key`{.str}, `value`{.str}, `opts`{.tbl .opt}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Stores the value under the key. Type is auto-detected: strings become bytes,
integers that fit in int64 become integers, other numbers become floats.
Pass `opts.ttl` (seconds) to set an expiration time.
]===]
-- Shared tail of the ks:set* family: TTL, optional encryption, one write txn.
local ks_set_packed = function(self, db, key, val_str, type_tag, opts)
	local ttl, terr = calc_ttl(opts)
	if not ttl then
		return nil, terr
	end
	if self.__state.subkey then
		val_str = enc_value(self.__state.subkey, key, val_str, type_tag, ttl)
	end
	return with_write_txn(db, function(txn)
		local ok, werr = ks_write(self.cfg.name, txn, key, val_str, type_tag, ttl)
		if not ok then
			return nil, werr
		end
		return true, true
	end)
end

local ks_set = function(self, key, value, opts)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end
	local val_str, type_tag, pack_err = pack_value(value)
	if not val_str then
		return nil, pack_err
	end
	return ks_set_packed(self, db, key, val_str, type_tag, opts)
end

---! Set a key-value pair with explicit bytes type
---@ ks:set_bytes(`key`{.str}, `value`{.str}, `opts`{.tbl .opt}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local ks_set_bytes = function(self, key, value, opts)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end
	return ks_set_packed(self, db, key, tostring(value), TYPE_BYTES, opts)
end

---! Set a key-value pair with explicit integer type
---@ ks:set_integer(`key`{.str}, `value`{.num}, `opts`{.tbl .opt}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local ks_set_integer = function(self, key, value, opts)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end
	-- NaN/inf/fractions would hit an undefined (int64_t) cast in C, and
	-- doubles past 2^53 silently lose integer precision.
	if type(value) ~= "number" or math.floor(value) ~= value or value < -2 ^ 53 or value > 2 ^ 53 then
		return nil, "value must be an integer within +/-2^53"
	end
	return ks_set_packed(self, db, key, core.pack_int64(value), TYPE_INTEGER, opts)
end

---! Set a key-value pair with explicit float type
---@ ks:set_float(`key`{.str}, `value`{.num}, `opts`{.tbl .opt}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local ks_set_float = function(self, key, value, opts)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end
	if type(value) ~= "number" then
		return nil, "value must be a number"
	end
	return ks_set_packed(self, db, key, core.pack_double(value), TYPE_FLOAT, opts)
end

---! Delete a key
---@ ks:del(`key`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local ks_del = function(self, key)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end

	return with_write_txn(db, function(txn)
		local ok, werr = ks_delete(self.cfg.name, txn, key)
		if not ok then
			-- A keyspace that was never written to has no directory entry;
			-- deleting from it is as much a no-op as deleting a missing key.
			if werr == "not found" then
				return true, true
			end
			return nil, werr
		end
		return true, true
	end)
end

---! Check if a key exists
---@ ks:exists(`key`{.str}) -> `exists`{.bool}
local ks_exists = function(self, key)
	local val = self:get(key)
	return val ~= nil
end

-- ── Integer operations ────────────────────────────────────────────────

---! Increment an integer key
---@ ks:incr(`key`{.str}, `delta`{.num .opt}) -> `new_value`{.num .or_nil}, `err`{.str .err}
--[===[
Atomically increments the integer stored at key by delta (default 1).
If the key does not exist, it is initialized to 0 before incrementing.
Returns the new value. Fails with "type mismatch" if the key holds a
non-integer type.
]===]
local ks_incr = function(self, key, delta)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end
	local d = delta or 1
	if type(d) ~= "number" or math.floor(d) ~= d or d < -2 ^ 53 or d > 2 ^ 53 then
		return nil, "delta must be an integer within +/-2^53"
	end

	return with_write_txn(db, function(txn)
		-- Read current value
		local raw, tag, ttl = ks_read(txn, self.cfg.name, key)
		local current = 0
		local preserved_ttl = 0
		if raw then
			-- Expiry first: an expired key behaves as non-existent
			-- regardless of the type it used to hold.
			if ttl and ttl ~= 0 and ttl < os.time() then
				current = 0
			elseif tag ~= TYPE_INTEGER then
				return nil, "type mismatch"
			else
				preserved_ttl = ttl or 0
				-- Decrypt if encrypted keyspace
				local val_raw = raw
				if self.__state.subkey then
					local dec, derr = dec_value(self.__state.subkey, key, raw, TYPE_INTEGER, ttl)
					if not dec then
						return nil, derr
					end
					val_raw = dec
				end
				current = core.unpack_int64(val_raw)
				if not current then
					return nil, "corrupt integer value"
				end
			end
		end

		local new_val = current + d

		local val_str = core.pack_int64(new_val)
		if self.__state.subkey then
			val_str = enc_value(self.__state.subkey, key, val_str, TYPE_INTEGER, preserved_ttl)
		end
		local ok, werr = ks_write(self.cfg.name, txn, key, val_str, TYPE_INTEGER, preserved_ttl)
		if not ok then
			return nil, werr
		end
		return true, new_val
	end)
end

---! Decrement an integer key
---@ ks:decr(`key`{.str}, `delta`{.num .opt}) -> `new_value`{.num .or_nil}, `err`{.str .err}
local ks_decr = function(self, key, delta)
	return self:incr(key, -(delta or 1))
end

-- ── TTL operations ────────────────────────────────────────────────────

---! Get remaining TTL for a key
---@ ks:ttl(`key`{.str}) -> `seconds`{.num .or_nil}
--[===[
Returns the remaining time-to-live in seconds, or nil if the key has no
TTL set, does not exist, or has already expired.
]===]
local ks_ttl = function(self, key)
	local db, cerr = check_key(self, key)
	if not db then
		return nil, cerr
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local raw, tag, ttl = ks_read(txn, self.cfg.name, key)
	txn:abort()

	if not raw then
		return nil
	end
	if not ttl or ttl == 0 then
		return nil
	end -- no TTL set

	-- Same boundary as get()/purge_expired(): the key is alive through
	-- its expiry second, so remaining can legitimately be 0.
	local now = os.time()
	if ttl < now then
		return nil
	end -- expired
	return ttl - now
end

---! Remove the TTL from a key
---@ ks:persist(`key`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local ks_persist = function(self, key)
	local db, cerr = check_write_key(self, key)
	if not db then
		return nil, cerr
	end

	return with_write_txn(db, function(txn)
		-- Read current value and re-write with ttl=0
		local raw, tag, ttl = ks_read(txn, self.cfg.name, key)
		if not raw then
			return nil, "not found"
		end
		-- Treat already-expired keys as not found (don't resurrect them).
		if ttl and ttl ~= 0 and ttl < os.time() then
			return nil, "not found"
		end

		-- On an encrypted keyspace the TTL is part of the AAD, so changing
		-- it requires re-encrypting under the new TTL=0; the stored
		-- ciphertext cannot simply be re-written with a different TTL.
		if self.__state.subkey then
			local dec, derr = dec_value(self.__state.subkey, key, raw, tag, ttl)
			if not dec then
				return nil, derr
			end
			raw = enc_value(self.__state.subkey, key, dec, tag, 0)
		end

		local ok, werr = ks_write(self.cfg.name, txn, key, raw, tag, 0)
		if not ok then
			return nil, werr
		end
		return true, true
	end)
end

---! Remove all expired keys from the keyspace
---@ ks:purge_expired() -> `count`{.num .or_nil}, `err`{.str .err}
local ks_purge_expired = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	-- First pass: collect expired keys (read-only)
	local txn_r, err = db.__state.handle:txn_begin(true)
	if not txn_r then
		return nil, err
	end

	local dir_root = txn_r:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn_r, dir_root, self.cfg.name)
	if not ks_root then
		txn_r:abort()
		return 0
	end

	local cur = txn_r:cursor_open(ks_root)
	if not cur then
		txn_r:abort()
		return 0
	end

	local expired = {}
	local now = os.time()
	local k, v, tag, ttl = cur:first()
	while k do
		if ttl and ttl ~= 0 and ttl < now then
			table.insert(expired, k)
		end
		k, v, tag, ttl = cur:next()
	end
	cur:close()
	txn_r:abort()

	if #expired == 0 then
		return 0
	end

	-- Second pass: delete expired keys (write), rechecking each key to
	-- guard against a concurrent writer refreshing them between the two phases.
	local txn_w, werr = db.__state.handle:txn_begin(false)
	if not txn_w then
		return nil, werr
	end

	local deleted = 0
	local now2 = os.time()
	for _, ek in ipairs(expired) do
		local raw, _, kttl = ks_read(txn_w, self.cfg.name, ek)
		-- Only delete if the key still exists and is still expired.
		if raw and kttl and kttl ~= 0 and kttl < now2 then
			local dok, derr = ks_delete(self.cfg.name, txn_w, ek)
			if not dok then
				txn_w:abort()
				return nil, derr
			end
			deleted = deleted + 1
		end
	end

	local cok, cerr = txn_w:commit()
	if not cok then
		return nil, cerr
	end
	return deleted
end

-- ── Scan / Range / Count ──────────────────────────────────────────────

local SCAN_BATCH_SIZE = 64

-- Shared iterator body for scan (prefix-bounded) and range (end-key
-- bounded): batched cursor reads, expired keys skipped, transparent
-- decryption with corrupted entries silently dropped.
local make_kv_iterator = function(self, start_key, end_key, prefix)
	local db = self.__state.db
	local subkey = self.__state.subkey -- nil if not encrypted
	local empty = function()
		return nil
	end
	if db.__state.closed then
		return empty
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return empty
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.cfg.name)
	if not ks_root then
		txn:abort()
		return empty
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return empty
	end

	if start_key and #start_key > 0 then
		cur:seek(start_key)
	else
		cur:first()
	end

	local done = false
	local now = os.time()
	local batch, batch_n, bi = nil, 0, 1

	return function()
		while not done do
			-- Refill batch
			if bi > batch_n * 4 then
				local berr
				batch, batch_n, berr = cur:next_batch(SCAN_BATCH_SIZE, end_key, prefix)
				bi = 1
				if not batch then
					done = true
					cur:close()
					txn:abort()
					error(berr or "cursor batch read failed")
				end
				if batch_n == 0 then
					done = true
					cur:close()
					txn:abort()
					return nil
				end
			end

			local k = batch[bi]
			local v = batch[bi + 1]
			local tag = batch[bi + 2]
			local ttl = batch[bi + 3]
			bi = bi + 4

			-- Skip expired keys and reserved internal records (\x00 prefix,
			-- e.g. the vector dimension key); user keys never start with NUL.
			if (ttl and ttl ~= 0 and ttl < now) or k:byte(1) == 0 then
				-- continue to next
			elseif subkey then
				local dec = dec_value(subkey, k, v, tag, ttl)
				if dec then
					return k, unpack_value(dec, tag)
				end
				-- decryption failed: skip corrupted entry
			else
				return k, unpack_value(v, tag)
			end
		end
	end
end

---! Iterate keys by prefix
---@ ks:scan(`prefix`{.str .opt}) -> `iterator`{.func}
--[===[
Returns a stateful iterator that yields `(key, value)` pairs for all keys
matching the prefix (or all keys if prefix is omitted). Keys are returned
in lexicographic order. Expired keys are skipped.
]===]
local ks_scan = function(self, prefix)
	local pfx = (prefix and #prefix > 0) and prefix or nil
	return make_kv_iterator(self, pfx, nil, pfx)
end

---! Iterate keys in a lexicographic range
---@ ks:range(`start_key`{.str .opt}, `end_key`{.str .opt}) -> `iterator`{.func}
--[===[
Returns a stateful iterator that yields `(key, value)` pairs for keys
in the half-open range `[start_key, end_key)`. Both bounds are optional:
nil start begins from the first key, nil end continues to the last.
]===]
local ks_range = function(self, start_key, end_key)
	return make_kv_iterator(self, start_key, end_key, nil)
end

---! Count keys matching a prefix
---@ ks:count(`prefix`{.str .opt}) -> `n`{.num}
local ks_count = function(self, prefix)
	local db = self.__state.db
	if db.__state.closed then
		return 0
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return 0
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.cfg.name)
	if not ks_root then
		txn:abort()
		return 0
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return 0
	end

	local pfx = (prefix and #prefix > 0) and prefix or nil
	if pfx then
		cur:seek(pfx)
	else
		cur:first()
	end

	-- Count non-expired entries without decrypting values (unlike scan)
	local n = 0
	local now = os.time()
	while true do
		local batch, batch_n = cur:next_batch(SCAN_BATCH_SIZE, nil, pfx)
		if not batch or batch_n == 0 then
			break
		end
		for i = 0, batch_n - 1 do
			local k = batch[i * 4 + 1]
			local ttl = batch[i * 4 + 4]
			-- Skip expired and reserved internal records (\x00 prefix).
			if not (ttl and ttl ~= 0 and ttl < now) and k:byte(1) ~= 0 then
				n = n + 1
			end
		end
	end
	cur:close()
	txn:abort()
	return n
end

-- ── Keyspace statistics ───────────────────────────────────────────────

---! Get B-tree statistics for the keyspace
---@ ks:stats() -> `stats`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a table with `keys`, `depth`, `leaf_pages`, `branch_pages`, and
`overflow_pages` fields.
]===]
local ks_stats = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.cfg.name)
	if not ks_root then
		txn:abort()
		return { keys = 0, depth = 0, leaf_pages = 0, branch_pages = 0, overflow_pages = 0 }
	end

	local stats = txn:btree_stats(ks_root)
	txn:abort()
	return stats
end

-- ── Batch operations ──────────────────────────────────────────────────

---! Execute multiple operations in a single transaction
---@ ks:batch(`fn`{.func}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Calls `fn(batch)` where the batch object supports `set`, `set_integer`,
`set_float`, `del`, and `incr` methods. All operations share a single
write transaction and are committed with one fsync on success. If fn
raises an error, the transaction is aborted.
]===]
local ks_batch = function(self, fn)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local txn, err = db.__state.handle:txn_begin(false)
	if not txn then
		return nil, err
	end

	-- Build batch object with same interface.
	-- All ops `error()` on failure so the surrounding pcall aborts the txn:
	-- a callback that ignores per-op errors must not be able to commit a
	-- half-applied batch.
	local sk = self.__state.subkey -- nil if not encrypted
	local check_batch_key = function(key)
		if type(key) ~= "string" then
			error("key must be a string", 0)
		end
		if #key == 0 then
			error("key must not be empty", 0)
		end
		if #key > MAX_KEY_LEN then
			error("key too long", 0)
		end
		if key:find("\x00", 1, true) then
			error("key must not contain null bytes", 0)
		end
	end
	local batch_ttl = function(opts)
		local ttl, terr = calc_ttl(opts)
		if not ttl then
			error(terr, 0)
		end
		return ttl
	end
	local batch = {
		set = function(_, key, value, opts)
			check_batch_key(key)
			local ttl = batch_ttl(opts)
			local val_str, type_tag, pack_err = pack_value(value)
			if not val_str then
				error(pack_err, 0)
			end
			if sk then
				val_str = enc_value(sk, key, val_str, type_tag, ttl)
			end
			local ok, werr = ks_write(self.cfg.name, txn, key, val_str, type_tag, ttl)
			if not ok then
				error(werr, 0)
			end
			return ok
		end,
		set_integer = function(_, key, value, opts)
			check_batch_key(key)
			local ttl = batch_ttl(opts)
			if type(value) ~= "number" or math.floor(value) ~= value or value < -2 ^ 53 or value > 2 ^ 53 then
				error("value must be an integer within +/-2^53", 0)
			end
			local val_str = core.pack_int64(value)
			if sk then
				val_str = enc_value(sk, key, val_str, TYPE_INTEGER, ttl)
			end
			local ok, werr = ks_write(self.cfg.name, txn, key, val_str, TYPE_INTEGER, ttl)
			if not ok then
				error(werr, 0)
			end
			return ok
		end,
		set_float = function(_, key, value, opts)
			check_batch_key(key)
			local ttl = batch_ttl(opts)
			if type(value) ~= "number" then
				error("value must be a number", 0)
			end
			local val_str = core.pack_double(value)
			if sk then
				val_str = enc_value(sk, key, val_str, TYPE_FLOAT, ttl)
			end
			local ok, werr = ks_write(self.cfg.name, txn, key, val_str, TYPE_FLOAT, ttl)
			if not ok then
				error(werr, 0)
			end
			return ok
		end,
		del = function(_, key)
			check_batch_key(key)
			local ok, werr = ks_delete(self.cfg.name, txn, key)
			if not ok then
				error(werr, 0)
			end
			return ok
		end,
		incr = function(_, key, delta)
			check_batch_key(key)
			local d = delta or 1
			if type(d) ~= "number" or math.floor(d) ~= d or d < -2 ^ 53 or d > 2 ^ 53 then
				error("delta must be an integer within +/-2^53", 0)
			end
			local raw, tag, ttl = ks_read(txn, self.cfg.name, key)
			local current = 0
			local preserved_ttl = 0
			if raw then
				if ttl and ttl ~= 0 and ttl < os.time() then
					current = 0
				elseif tag ~= TYPE_INTEGER then
					error("type mismatch", 0)
				else
					preserved_ttl = ttl or 0
					local val_raw = raw
					if sk then
						local dec, derr = dec_value(sk, key, raw, TYPE_INTEGER, ttl)
						if not dec then
							error(derr, 0)
						end
						val_raw = dec
					end
					current = core.unpack_int64(val_raw)
					if not current then
						error("corrupt integer value", 0)
					end
				end
			end
			local new_val = current + d
			local val_str = core.pack_int64(new_val)
			if sk then
				val_str = enc_value(sk, key, val_str, TYPE_INTEGER, preserved_ttl)
			end
			local ok, werr = ks_write(self.cfg.name, txn, key, val_str, TYPE_INTEGER, preserved_ttl)
			if not ok then
				error(werr, 0)
			end
			return new_val
		end,
	}

	local ok, batch_err = pcall(fn, batch)
	if not ok then
		txn:abort()
		return nil, batch_err
	end

	local cok, cerr = txn:commit()
	if not cok then
		return nil, cerr
	end
	return true
end

-- ── Vector operations ─────────────────────────────────────────────────

-- The dimension metadata key VDIM_KEY (defined next to check_key, which
-- blocks plain access to it) is stored in the keyspace alongside vectors.

---! Store a vector
---@ ks:vset(`key`{.str}, `vec`{.tbl}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Stores a float32 vector (a Lua table of numbers). The dimension is inferred
from the table length and must be consistent across all vectors in the
keyspace. Not supported with encryption.
]===]
local ks_vset = function(self, key, vec_table)
	if self.__state.subkey then
		return nil, "encryption not supported for vectors"
	end
	if type(key) ~= "string" then
		return nil, "key must be a string"
	end
	if #key == 0 or #key > MAX_KEY_LEN then
		return nil, "invalid key length"
	end
	-- NUL keys would collide with the reserved dimension record (VDIM_KEY).
	if key:find("\x00", 1, true) then
		return nil, "key must not contain null bytes"
	end
	if type(vec_table) ~= "table" then
		return nil, "vector must be a table of numbers"
	end
	if #vec_table == 0 then
		return nil, "vector must not be empty"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local txn, err = db.__state.handle:txn_begin(false)
	if not txn then
		return nil, err
	end

	local ks_name = self.cfg.name
	local dim = #vec_table

	-- Type guard: never silently replace a non-vector value
	local existing_raw, existing_tag = ks_read(txn, ks_name, key)
	if existing_raw and existing_tag ~= TYPE_VECTOR then
		txn:abort()
		return nil, "type mismatch: key is not a vector"
	end

	-- Check dimension consistency
	local dim_raw = ks_read(txn, ks_name, VDIM_KEY)
	if dim_raw then
		local existing_dim = core.unpack_int64(dim_raw)
		if existing_dim ~= dim then
			txn:abort()
			return nil, "dimension mismatch"
		end
	else
		-- First vector: record dimension
		local ok, werr = ks_write(ks_name, txn, VDIM_KEY, core.pack_int64(dim), TYPE_INTEGER, 0)
		if not ok then
			txn:abort()
			return nil, werr
		end
	end

	-- Pack vector as [uint16 dim][float32 × dim]
	local packed = core.pack_floats(vec_table)
	if not packed then
		txn:abort()
		return nil, "pack failed"
	end

	-- Store with TYPE_VECTOR
	local ok, werr = ks_write(ks_name, txn, key, packed, TYPE_VECTOR, 0)
	if not ok then
		txn:abort()
		return nil, werr
	end

	local ok, cerr = txn:commit()
	if not ok then
		return nil, cerr
	end
	return true
end

---! Retrieve a vector
---@ ks:vget(`key`{.str}) -> `vec`{.tbl .or_nil}, `err`{.str .err}
local ks_vget = function(self, key)
	if self.__state.subkey then
		return nil, "encryption not supported for vectors"
	end
	if type(key) ~= "string" then
		return nil, "key must be a string"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local raw, tag = ks_read(txn, self.cfg.name, key)
	txn:abort()

	if not raw then
		return nil
	end
	if tag ~= TYPE_VECTOR then
		return nil, "not a vector"
	end

	return core.unpack_floats(raw)
end

---! Search for similar vectors
---@ ks:vsearch(`query`{.tbl}, `opts`{.tbl .opt}) -> `results`{.tbl}
--[===[
Returns an array of `{key, score}` tables sorted by similarity.
Options: `metric` ("cosine", "dot", or "l2"; default "cosine"),
`top_k` (max results; default 10), `filter` (key prefix filter).
Cosine: 0=identical, 2=opposite. Dot: higher=more similar. L2: 0=identical.
]===]
local ks_vsearch = function(self, query_vec, opts)
	if self.__state.subkey then
		return nil, "encryption not supported for vectors"
	end
	if type(query_vec) ~= "table" then
		return nil, "query must be a table"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local metric_name = (opts and opts.metric) or "cosine"
	local top_k = (opts and opts.top_k) or 10
	local filter = opts and opts.filter

	if type(top_k) ~= "number" or top_k ~= math.floor(top_k) or top_k < 1 or top_k > 10000 then
		return nil, "top_k must be a positive integer <= 10000"
	end

	local metric_id
	if metric_name == "cosine" then
		metric_id = METRIC_COSINE
	elseif metric_name == "dot" then
		metric_id = METRIC_DOT
	elseif metric_name == "l2" then
		metric_id = METRIC_L2
	else
		return nil, "unknown metric: " .. tostring(metric_name)
	end

	-- Pack query vector
	local q_packed = core.pack_floats(query_vec)
	if not q_packed then
		return nil, "pack failed"
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, "txn failed"
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.cfg.name)
	if not ks_root then
		txn:abort()
		return {}
	end

	-- The C scan silently skips vectors whose dimension differs from the
	-- query, so a wrong-dimension query would return {} with no diagnostic.
	local dim_raw = ks_read(txn, self.cfg.name, VDIM_KEY)
	if not dim_raw then
		txn:abort()
		return {}
	end
	if core.unpack_int64(dim_raw) ~= #query_vec then
		txn:abort()
		return nil, "dimension mismatch"
	end

	local results = core.vsearch_scan(txn, ks_root, q_packed, metric_id, top_k, filter)
	txn:abort()

	return results or {}
end

---! Get the vector dimension of the keyspace
---@ ks:vdim() -> `dim`{.num .or_nil}
local ks_vdim = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local raw, tag = ks_read(txn, self.cfg.name, VDIM_KEY)
	txn:abort()

	if not raw then
		return nil
	end
	if tag ~= TYPE_INTEGER then
		return nil, "invalid dimension record"
	end
	return core.unpack_int64(raw)
end

-- ── Sorted Set ────────────────────────────────────────────────────────

-- Composite key prefixes:
-- name \x00 \x00              → metadata (cardinality as int64)
-- name \x00 \x01 <score> \x00 member → score index
-- name \x00 \x02 member       → member index (value = score bytes)

local SEP = "\x00"

local zs_meta_key = function(name)
	return name .. SEP .. "\x00"
end
local zs_score_prefix = function(name)
	return name .. SEP .. "\x01"
end
local zs_member_prefix = function(name)
	return name .. SEP .. "\x02"
end

local zs_score_key = function(name, score_bytes, member)
	return name .. SEP .. "\x01" .. score_bytes .. SEP .. member
end
local zs_member_key = function(name, member)
	return name .. SEP .. "\x02" .. member
end

-- Worst-case composite key length for a (name, member) pair is the score-index
-- key: name + SEP + marker(1) + score(8) + SEP + member.  Reject early so the
-- caller gets "key too long" instead of a generic btree failure deep in C.
local zs_check_composite = function(name, member)
	if type(member) ~= "string" then
		return nil, "member must be a string"
	end
	if #name + 1 + 1 + 8 + 1 + #member > MAX_KEY_LEN then
		return nil, "key too long"
	end
	return true
end

-- Type guard: verify metadata belongs to a sorted set
local zs_check_type = function(txn, ks_name, name)
	local raw, tag = ks_read(txn, ks_name, zs_meta_key(name))
	if not raw then
		return true
	end
	if tag ~= TYPE_ZSET then
		return nil, "type mismatch: key is not a sorted set"
	end
	return true
end

---! Add or update members with scores
---@ zs:add(`score`{.num}, `member`{.str}, `...`{.opt}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Accepts one or more (score, member) pairs as varargs. If a member already
exists, its score is updated. New members increment the cardinality.
]===]
local zs_add = function(self, ...)
	local args = { ... }
	if #args < 2 or #args % 2 ~= 0 then
		return nil, "add requires (score, member) pairs"
	end

	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	for i = 1, #args, 2 do
		if type(args[i]) ~= "number" then
			return nil, "score must be a number"
		end
		local cok, cerr = zs_check_composite(self.cfg.name, args[i + 1])
		if not cok then
			return nil, cerr
		end
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		local tok, terr = zs_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		-- Read current cardinality
		local meta_raw = ks_read(txn, ks_name, zs_meta_key(name))
		local card = 0
		if meta_raw then
			card = core.unpack_int64(meta_raw)
		end

		for i = 1, #args, 2 do
			local score = args[i]
			local member = args[i + 1]
			local score_bytes = core.encode_score(score)

			-- Check if member already exists (for update)
			local old_score_raw = ks_read(txn, ks_name, zs_member_key(name, member))
			if old_score_raw then
				-- Remove old score index entry; a silently ignored failure
				-- here would commit two score entries for one member.
				local dok, derr = ks_delete(ks_name, txn, zs_score_key(name, old_score_raw, member))
				if not dok then
					return nil, derr
				end
			else
				card = card + 1
			end

			-- Write score index entry (empty value)
			local ok, werr = ks_write(ks_name, txn, zs_score_key(name, score_bytes, member), "", TYPE_BYTES, 0)
			if not ok then
				return nil, werr
			end

			-- Write member index entry (value = score_bytes)
			ok, werr = ks_write(ks_name, txn, zs_member_key(name, member), score_bytes, TYPE_BYTES, 0)
			if not ok then
				return nil, werr
			end
		end

		-- Update cardinality
		local ok, werr = ks_write(ks_name, txn, zs_meta_key(name), core.pack_int64(card), TYPE_ZSET, 0)
		if not ok then
			return nil, werr
		end
		return true, true
	end)
end

---! Remove a member from the sorted set
---@ zs:rem(`member`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local zs_rem = function(self, member)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local cok, cerr = zs_check_composite(self.cfg.name, member)
	if not cok then
		return nil, cerr
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		local tok, terr = zs_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		-- Look up member's score
		local score_raw = ks_read(txn, ks_name, zs_member_key(name, member))
		if not score_raw then
			return nil, "not found"
		end

		-- Delete score index entry
		local dok, derr = ks_delete(ks_name, txn, zs_score_key(name, score_raw, member))
		if not dok then
			return nil, derr
		end
		-- Delete member index entry
		dok, derr = ks_delete(ks_name, txn, zs_member_key(name, member))
		if not dok then
			return nil, derr
		end

		-- Decrement cardinality
		local meta_raw = ks_read(txn, ks_name, zs_meta_key(name))
		local card = meta_raw and core.unpack_int64(meta_raw) or 0
		if card > 0 then
			card = card - 1
		end
		local wok, werr = ks_write(ks_name, txn, zs_meta_key(name), core.pack_int64(card), TYPE_ZSET, 0)
		if not wok then
			return nil, werr
		end

		return true, true
	end)
end

---! Get the score of a member
---@ zs:score(`member`{.str}) -> `score`{.num .or_nil}
local zs_score = function(self, member)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local cok = zs_check_composite(self.cfg.name, member)
	if not cok then
		return nil
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return nil
	end

	local tok, terr = zs_check_type(txn, self.__state.ks_name, self.cfg.name)
	if not tok then
		txn:abort()
		return nil, terr
	end

	local score_raw = ks_read(txn, self.__state.ks_name, zs_member_key(self.cfg.name, member))
	txn:abort()

	if not score_raw then
		return nil
	end
	return core.decode_score(score_raw)
end

---! Get the cardinality of the sorted set
---@ zs:card() -> `count`{.num}
local zs_card = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return 0
	end

	local meta_raw, tag = ks_read(txn, self.__state.ks_name, zs_meta_key(self.cfg.name))
	txn:abort()

	if not meta_raw then
		return 0
	end
	if tag ~= TYPE_ZSET then
		return nil, "type mismatch: key is not a sorted set"
	end
	return core.unpack_int64(meta_raw)
end

---! Get the rank of a member in ascending score order
---@ zs:rank(`member`{.str}) -> `rank`{.num .or_nil}
local zs_rank = function(self, member)
	local cok = zs_check_composite(self.cfg.name, member)
	if not cok then
		return nil
	end
	local score = self:score(member)
	if not score then
		return nil
	end
	local score_bytes = core.encode_score(score)
	local target_key = zs_score_key(self.cfg.name, score_bytes, member)

	local db = self.__state.db
	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return nil
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.__state.ks_name)
	if not ks_root then
		txn:abort()
		return nil
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return nil
	end

	-- O(n) walk of the score index, batched to limit Lua<->C crossings
	local prefix = zs_score_prefix(self.cfg.name)
	local rank = 0
	cur:seek(prefix)
	while true do
		local batch, batch_n = cur:next_batch(SCAN_BATCH_SIZE, nil, prefix)
		if not batch or batch_n == 0 then
			break
		end
		for i = 0, batch_n - 1 do
			if batch[i * 4 + 1] == target_key then
				cur:close()
				txn:abort()
				return rank
			end
			rank = rank + 1
		end
	end
	cur:close()
	txn:abort()
	return nil
end

---! Get members within a score range
---@ zs:range_by_score(`min`{.num}, `max`{.num}, `opts`{.tbl .opt}) -> `results`{.tbl}
--[===[
Returns an array of `{score, member}` tables for members with scores in
`[min, max]`. Options: `limit` (max results), `reverse` (descending order;
when true, pass (high, low) as min/max).
]===]
local zs_range_by_score = function(self, min_score, max_score, opts)
	if type(min_score) ~= "number" or type(max_score) ~= "number" then
		return nil, "min and max must be numbers"
	end
	local limit = opts and opts.limit
	local reverse = opts and opts.reverse
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return {}
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.__state.ks_name)
	if not ks_root then
		txn:abort()
		return {}
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return {}
	end

	local name = self.cfg.name
	local prefix = zs_score_prefix(name)
	local results = {}

	-- For reverse, caller passes (high, low) so swap
	local lo, hi = min_score, max_score
	if reverse and lo > hi then
		lo, hi = hi, lo
	end

	if not reverse then
		local min_bytes = core.encode_score(lo)
		local max_bytes = core.encode_score(hi)
		local seek_key = prefix .. min_bytes

		local k = cur:seek(seek_key)
		while k do
			if k:sub(1, #prefix) ~= prefix then
				break
			end
			-- Extract score bytes (8 bytes after prefix)
			local sb = k:sub(#prefix + 1, #prefix + 8)
			if sb > max_bytes then
				break
			end
			-- Extract member (after score bytes + \x00 separator)
			local member = k:sub(#prefix + 8 + 1 + 1) -- +1 for \x00, +1 for 1-based
			table.insert(results, { score = core.decode_score(sb), member = member })
			if limit and #results >= limit then
				break
			end
			k = cur:next()
		end
	else
		local max_bytes = core.encode_score(hi)
		local min_bytes = core.encode_score(lo)
		-- For reverse: seek to a key just past max_score entries, then go backward.
		-- The score key format is: prefix + score_bytes(8) + \x00 + member
		-- A key with max_bytes + \xff will sort after all entries with max_bytes.
		local seek_key = prefix .. max_bytes .. "\xff"
		local k = cur:seek(seek_key)
		if k then
			-- seek landed on or after our target, step back
			k = cur:prev()
		else
			-- past end of tree, start from the very last key
			k = cur:last()
		end
		while k do
			if k:sub(1, #prefix) ~= prefix then
				break
			end
			local sb = k:sub(#prefix + 1, #prefix + 8)
			if sb < min_bytes then
				break
			end
			local member = k:sub(#prefix + 8 + 1 + 1)
			table.insert(results, { score = core.decode_score(sb), member = member })
			if limit and #results >= limit then
				break
			end
			k = cur:prev()
		end
	end

	cur:close()
	txn:abort()
	return results
end

---! Get members by rank range
---@ zs:range_by_rank(`start`{.num}, `stop`{.num}) -> `results`{.tbl}
--[===[
Returns an array of `{score, member}` tables for members at ranks in
`[start, stop]` (0-based, inclusive).
]===]
local zs_range_by_rank = function(self, start_rank, stop_rank)
	local s = start_rank or 0
	local e = stop_rank or -1
	if type(s) ~= "number" or type(e) ~= "number" then
		return nil, "start and stop must be numbers"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return {}
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.__state.ks_name)
	if not ks_root then
		txn:abort()
		return {}
	end

	-- Normalize negative ranks against the cardinality (Redis semantics:
	-- -1 is the last member), so range_by_rank(0, -1) returns the full set.
	if s < 0 or e < 0 then
		local meta_raw = ks_read(txn, self.__state.ks_name, zs_meta_key(self.cfg.name))
		local card = meta_raw and core.unpack_int64(meta_raw) or 0
		if s < 0 then
			s = card + s
		end
		if e < 0 then
			e = card + e
		end
	end
	if s < 0 then
		s = 0
	end
	if e < s then
		txn:abort()
		return {}
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return {}
	end

	local prefix = zs_score_prefix(self.cfg.name)
	local results = {}
	local rank = 0

	local k = cur:seek(prefix)
	while k do
		if k:sub(1, #prefix) ~= prefix then
			break
		end
		if rank >= s and rank <= e then
			local sb = k:sub(#prefix + 1, #prefix + 8)
			local member = k:sub(#prefix + 8 + 1 + 1)
			table.insert(results, { score = core.decode_score(sb), member = member })
		end
		if rank > e then
			break
		end
		rank = rank + 1
		k = cur:next()
	end

	cur:close()
	txn:abort()
	return results
end

---! Iterate all members in score order
---@ zs:iter(`opts`{.tbl .opt}) -> `iterator`{.func}
--[===[
Returns a stateful iterator that yields `(score, member)` pairs in
ascending score order. Pass `opts.reverse = true` for descending order.
]===]
local zs_iter = function(self, opts)
	local reverse = opts and opts.reverse
	local min_bytes = (opts and type(opts.min) == "number") and core.encode_score(opts.min) or nil
	local max_bytes = (opts and type(opts.max) == "number") and core.encode_score(opts.max) or nil
	local db = self.__state.db
	if db.__state.closed then
		return function()
			return nil
		end
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return function()
			return nil
		end
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.__state.ks_name)
	if not ks_root then
		txn:abort()
		return function()
			return nil
		end
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return function()
			return nil
		end
	end

	local prefix = zs_score_prefix(self.cfg.name)
	local started = false
	local done = false

	return function()
		if done then
			return nil
		end
		local k
		if not started then
			started = true
			if not reverse then
				-- With a min bound, start at the first entry >= min.
				k = cur:seek(min_bytes and (prefix .. min_bytes) or prefix)
			else
				-- Seek to the exclusive upper bound of the score-index range.
				-- With a max bound that is prefix..max..\xff (sorts after every
				-- member at max); otherwise the member-index prefix (marker
				-- \x02 sorts after every score key, marker \x01, of this set).
				local past = max_bytes and (prefix .. max_bytes .. "\xff") or zs_member_prefix(self.cfg.name)
				k = cur:seek(past)
				if k then
					k = cur:prev()
				else
					k = cur:last()
				end
			end
		else
			if not reverse then
				k = cur:next()
			else
				k = cur:prev()
			end
		end
		if not k or k:sub(1, #prefix) ~= prefix then
			done = true
			cur:close()
			txn:abort()
			return nil
		end
		local sb = k:sub(#prefix + 1, #prefix + 8)
		-- Score bounds are inclusive in both directions.
		if (not reverse and max_bytes and sb > max_bytes) or (reverse and min_bytes and sb < min_bytes) then
			done = true
			cur:close()
			txn:abort()
			return nil
		end
		local member = k:sub(#prefix + 8 + 1 + 1)
		return core.decode_score(sb), member
	end
end

---! Increment scores for multiple members in one transaction
---@ zs:batch_incr(`members`{.tbl}, `delta`{.num .opt}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Atomically increments the score of each member in the array by delta
(default 1). New members are created with score equal to delta.
Uses a single transaction with one fdatasync.
]===]
local zs_batch_incr = function(self, members, delta)
	if not members or #members == 0 then
		return true
	end
	local d = delta or 1
	if type(d) ~= "number" then
		return nil, "delta must be a number"
	end

	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	for _, member in ipairs(members) do
		local cok, cerr = zs_check_composite(name, member)
		if not cok then
			return nil, cerr
		end
	end

	return with_write_txn(db, function(txn)
		local tok, terr = zs_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		local meta_raw = ks_read(txn, ks_name, zs_meta_key(name))
		local card = meta_raw and core.unpack_int64(meta_raw) or 0

		for _, member in ipairs(members) do
			local old_score_raw = ks_read(txn, ks_name, zs_member_key(name, member))
			local new_score
			if old_score_raw then
				new_score = core.decode_score(old_score_raw) + d
				local dok, derr = ks_delete(ks_name, txn, zs_score_key(name, old_score_raw, member))
				if not dok then
					return nil, derr
				end
			else
				new_score = d
				card = card + 1
			end

			local score_bytes = core.encode_score(new_score)
			local ok, werr = ks_write(ks_name, txn, zs_score_key(name, score_bytes, member), "", TYPE_BYTES, 0)
			if not ok then
				return nil, werr
			end
			ok, werr = ks_write(ks_name, txn, zs_member_key(name, member), score_bytes, TYPE_BYTES, 0)
			if not ok then
				return nil, werr
			end
		end

		local mok, merr = ks_write(ks_name, txn, zs_meta_key(name), core.pack_int64(card), TYPE_ZSET, 0)
		if not mok then
			return nil, merr
		end
		return true, true
	end)
end

local new_sorted_set = function(db, ks_name, name)
	return {
		cfg = { name = name },
		__state = { db = db, ks_name = ks_name },
		add = zs_add,
		rem = zs_rem,
		score = zs_score,
		card = zs_card,
		rank = zs_rank,
		range_by_score = zs_range_by_score,
		range_by_rank = zs_range_by_rank,
		iter = zs_iter,
		batch_incr = zs_batch_incr,
	}
end

---! Get a sorted set handle within this keyspace
---@ ks:sorted_set(`name`{.str}) -> `zset`{.tbl}
local ks_sorted_set = function(self, name)
	if type(name) ~= "string" or #name == 0 then
		return nil, "sorted set name must be a non-empty string"
	end
	-- NUL is the composite-key separator: a name containing it could alias
	-- another structure's metadata or element keys.
	if name:find("\x00", 1, true) then
		return nil, "sorted set name must not contain null bytes"
	end
	-- Worst-case key footprint: SEP + marker(1) + score(8) + SEP + member.
	-- Leave at least one byte for the member; cap name at MAX_KEY_LEN - 11.
	if #name > MAX_KEY_LEN - 11 then
		return nil, "sorted set name too long"
	end
	return new_sorted_set(self.__state.db, self.cfg.name, name)
end

-- ── List ──────────────────────────────────────────────────────────────

-- Storage:
-- name \x00 \x00                → metadata (head_seq, tail_seq, length as 3x int64 = 24 bytes)
-- name \x00 <seq_bytes_be>      → element value

local SEQ_MID = 2 ^ 48 -- Starting point (within double exact integer range, ~2.8e14 ops each direction)

local list_meta_key = function(name)
	return name .. SEP .. "\x00"
end
local list_elem_key = function(name, seq)
	return name .. SEP .. core.pack_int64_be(seq)
end

-- Type guard: verify metadata belongs to a list
local list_check_type = function(txn, ks_name, name)
	local raw, tag = ks_read(txn, ks_name, list_meta_key(name))
	if not raw then
		return true
	end
	if tag ~= TYPE_LIST then
		return nil, "type mismatch: key is not a list"
	end
	return true
end

local list_read_meta = function(txn, ks_name, name)
	local raw = ks_read(txn, ks_name, list_meta_key(name))
	if not raw or #raw < 24 then
		return SEQ_MID, SEQ_MID - 1, 0 -- head, tail, length (empty: head > tail)
	end
	local head = core.unpack_int64(raw:sub(1, 8))
	local tail = core.unpack_int64(raw:sub(9, 16))
	local length = core.unpack_int64(raw:sub(17, 24))
	return head, tail, length
end

local list_write_meta = function(db, ks_name, txn, name, head, tail, length)
	local raw = core.pack_int64(head) .. core.pack_int64(tail) .. core.pack_int64(length)
	return ks_write(ks_name, txn, list_meta_key(name), raw, TYPE_LIST, 0)
end

---! Push a value to the right (tail) of the list
---@ lst:rpush(`value`{.str}) -> `length`{.num .or_nil}, `err`{.str .err}
local lst_rpush = function(self, value)
	if type(value) ~= "string" and type(value) ~= "number" then
		return nil, "value must be a string or number"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		local tok, terr = list_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		local head, tail, length = list_read_meta(txn, ks_name, name)
		tail = tail + 1
		length = length + 1

		local elem_key = list_elem_key(name, tail)
		local val_str = tostring(value)
		if self.__state.subkey then
			-- List elements are always stored as (TYPE_BYTES, ttl=0).
			val_str = enc_value(self.__state.subkey, elem_key, val_str, TYPE_BYTES, 0)
		end
		local ok, werr = ks_write(ks_name, txn, elem_key, val_str, TYPE_BYTES, 0)
		if not ok then
			return nil, werr
		end
		ok, werr = list_write_meta(db, ks_name, txn, name, head, tail, length)
		if not ok then
			return nil, werr
		end
		return true, length
	end)
end

---! Push a value to the left (head) of the list
---@ lst:lpush(`value`{.str}) -> `length`{.num .or_nil}, `err`{.str .err}
local lst_lpush = function(self, value)
	if type(value) ~= "string" and type(value) ~= "number" then
		return nil, "value must be a string or number"
	end
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		local tok, terr = list_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		local head, tail, length = list_read_meta(txn, ks_name, name)
		head = head - 1
		length = length + 1

		local elem_key = list_elem_key(name, head)
		local val_str = tostring(value)
		if self.__state.subkey then
			-- List elements are always stored as (TYPE_BYTES, ttl=0).
			val_str = enc_value(self.__state.subkey, elem_key, val_str, TYPE_BYTES, 0)
		end
		local ok, werr = ks_write(ks_name, txn, elem_key, val_str, TYPE_BYTES, 0)
		if not ok then
			return nil, werr
		end
		ok, werr = list_write_meta(db, ks_name, txn, name, head, tail, length)
		if not ok then
			return nil, werr
		end

		return true, length
	end)
end

---! Pop a value from the right (tail) of the list
---@ lst:rpop() -> `value`{.str .or_nil}
local lst_rpop = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		local tok, terr = list_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		local head, tail, length = list_read_meta(txn, ks_name, name)
		if length <= 0 then
			return false, nil
		end

		local elem_key = list_elem_key(name, tail)
		local val = ks_read(txn, ks_name, elem_key)
		if val and self.__state.subkey then
			local dec, derr = dec_value(self.__state.subkey, elem_key, val, TYPE_BYTES, 0)
			if not dec then
				-- Fail closed: do not delete or return ciphertext.
				return nil, derr
			end
			val = dec
		end
		local dok, derr = ks_delete(ks_name, txn, elem_key)
		if not dok then
			return nil, derr
		end
		tail = tail - 1
		length = length - 1
		local ok, werr = list_write_meta(db, ks_name, txn, name, head, tail, length)
		if not ok then
			return nil, werr
		end

		return true, val
	end)
end

---! Pop a value from the left (head) of the list
---@ lst:lpop() -> `value`{.str .or_nil}
local lst_lpop = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		local tok, terr = list_check_type(txn, ks_name, name)
		if not tok then
			return nil, terr
		end

		local head, tail, length = list_read_meta(txn, ks_name, name)
		if length <= 0 then
			return false, nil
		end

		local elem_key = list_elem_key(name, head)
		local val = ks_read(txn, ks_name, elem_key)
		if val and self.__state.subkey then
			local dec, derr = dec_value(self.__state.subkey, elem_key, val, TYPE_BYTES, 0)
			if not dec then
				-- Fail closed: do not delete or return ciphertext.
				return nil, derr
			end
			val = dec
		end
		local dok, derr = ks_delete(ks_name, txn, elem_key)
		if not dok then
			return nil, derr
		end
		head = head + 1
		length = length - 1
		local ok, werr = list_write_meta(db, ks_name, txn, name, head, tail, length)
		if not ok then
			return nil, werr
		end

		return true, val
	end)
end

---! Get the length of the list
---@ lst:len() -> `length`{.num}
local lst_len = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return 0
	end

	local _, _, length = list_read_meta(txn, self.__state.ks_name, self.cfg.name)
	txn:abort()
	return length
end

---! Get element at index
---@ lst:index(`idx`{.num}) -> `value`{.str .or_nil}
--[===[
Returns the element at the given 0-based index. Negative indices count
from the end: -1 is the last element.
]===]
local lst_index = function(self, idx)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if type(idx) ~= "number" then
		return nil, "index must be a number"
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return nil
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name
	local head, tail, length = list_read_meta(txn, ks_name, name)

	-- Handle negative indices
	local actual = idx
	if actual < 0 then
		actual = length + actual
	end
	if actual < 0 or actual >= length then
		txn:abort()
		return nil
	end

	local seq = head + actual
	local elem_key = list_elem_key(name, seq)
	local val = ks_read(txn, ks_name, elem_key)
	txn:abort()
	if val and self.__state.subkey then
		local dec, derr = dec_value(self.__state.subkey, elem_key, val, TYPE_BYTES, 0)
		if not dec then
			return nil, derr
		end
		val = dec
	end
	return val
end

---! Get a range of elements
---@ lst:range(`start`{.num .opt}, `stop`{.num .opt}) -> `items`{.tbl}
--[===[
Returns an array of values for elements at indices `[start, stop]`
(0-based, inclusive). Negative indices wrap from the end. Defaults:
start=0, stop=-1 (entire list).
]===]
local lst_range = function(self, start_idx, stop_idx)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return {}
	end

	local name = self.cfg.name
	local ks_name = self.__state.ks_name
	local head, tail, length = list_read_meta(txn, ks_name, name)

	-- Normalize indices (0-based, inclusive, negative wraps)
	local s = start_idx or 0
	local e = stop_idx or -1
	if s < 0 then
		s = length + s
	end
	if e < 0 then
		e = length + e
	end
	if s < 0 then
		s = 0
	end
	if e >= length then
		e = length - 1
	end

	local results = {}
	local corrupted = 0
	local sk = self.__state.subkey

	if e >= s then
		-- Element keys are contiguous big-endian sequence numbers: one
		-- cursor sweep instead of a full tree descent per element.
		local dir_root = txn:meta().keyspace_root_pgno
		local ks_root = ks_dir_get(txn, dir_root, ks_name)
		local cur = ks_root and txn:cursor_open(ks_root)
		if cur then
			cur:seek(list_elem_key(name, head + s))
			local end_key = list_elem_key(name, head + e + 1)
			while true do
				local batch, batch_n = cur:next_batch(SCAN_BATCH_SIZE, end_key)
				if not batch or batch_n == 0 then
					break
				end
				for i = 0, batch_n - 1 do
					local elem_key = batch[i * 4 + 1]
					local val = batch[i * 4 + 2]
					if sk then
						local dec = dec_value(sk, elem_key, val, TYPE_BYTES, 0)
						if dec then
							val = dec
						else
							val = nil
							corrupted = corrupted + 1
						end
					end
					if val then
						table.insert(results, val)
					end
				end
			end
			cur:close()
		end
	end

	txn:abort()
	if corrupted > 0 then
		return results, corrupted
	end
	return results
end

local new_list = function(db, ks_name, name, subkey)
	return {
		cfg = { name = name },
		__state = { db = db, ks_name = ks_name, subkey = subkey },
		rpush = lst_rpush,
		lpush = lst_lpush,
		rpop = lst_rpop,
		lpop = lst_lpop,
		len = lst_len,
		index = lst_index,
		range = lst_range,
	}
end

---! Get a list handle within this keyspace
---@ ks:list(`name`{.str}) -> `list`{.tbl}
local ks_list = function(self, name)
	if type(name) ~= "string" or #name == 0 then
		return nil, "list name must be a non-empty string"
	end
	-- NUL is the composite-key separator: a name containing it could alias
	-- another structure's metadata or element keys.
	if name:find("\x00", 1, true) then
		return nil, "list name must not contain null bytes"
	end
	-- Element key footprint: name + SEP + int64(8); cap name at MAX_KEY_LEN - 9.
	if #name > MAX_KEY_LEN - 9 then
		return nil, "list name too long"
	end
	return new_list(self.__state.db, self.cfg.name, name, self.__state.subkey)
end

-- ── Full-text validation helpers ──────────────────────────────────────

-- Maximum term length matches the default C analyzer truncation point.
local FT_MAX_TERM_LEN = 255

-- Validate a doc_id and ensure all FT composite keys for it fit within
-- MAX_KEY_LEN.  Returns true or nil, err.
local check_doc_id = function(ft_name, doc_id)
	if type(doc_id) ~= "string" or #doc_id == 0 then
		return nil, "doc_id must be a non-empty string"
	end
	if doc_id:find("\x00") then
		return nil, "doc_id must not contain null bytes"
	end
	-- Longest key using doc_id is ft_reverse_key: name .. "\x04" .. doc_id
	-- (ft_posting_key adds term + "\x00\x01" but terms are validated separately)
	if #ft_name + 1 + #doc_id > MAX_KEY_LEN then
		return nil, "doc_id too long"
	end
	return true
end

-- Validate an analyzer output term and check that its posting/df composite
-- keys fit within MAX_KEY_LEN.  Returns true or nil, err.
local check_ft_term = function(ft_name, term, doc_id)
	if type(term) ~= "string" or #term == 0 then
		return nil, "term must be a non-empty string"
	end
	if term:find("\x00") then
		return nil, "term must not contain null bytes"
	end
	if #term > FT_MAX_TERM_LEN then
		return nil, "term too long"
	end
	-- Longest key: ft_posting_key = name .. "\x03" .. term .. "\x00\x01" .. doc_id
	if #ft_name + 1 + #term + 2 + #doc_id > MAX_KEY_LEN then
		return nil, "key too long"
	end
	return true
end

-- ── Full-text search keyspace ─────────────────────────────────────────

-- FT key construction helpers
local ft_meta_key = function(name)
	return name .. "\x00\x00"
end

local ft_analyzer_key = function(name)
	return name .. "\x00\x01"
end

-- Present while bulk-load indexing is paused; cleared by rebuild_index().
-- Persisted so other handles -- and reopens after a crash mid-load -- see
-- the index as incomplete instead of searching it.
local ft_paused_key = function(name)
	return name .. "\x00\x02"
end

local ft_is_paused = function(self, txn)
	if self.__state.indexing_paused then
		return true
	end
	return ks_read(txn, self.__state.ks_name, ft_paused_key(self.cfg.name)) ~= nil
end

local ft_doc_key = function(name, doc_id)
	return name .. "\x01" .. doc_id
end

local ft_doc_meta_key = function(name, doc_id)
	return name .. "\x02" .. doc_id
end

local ft_df_key = function(name, term)
	return name .. "\x03" .. term .. "\x00\x00"
end

local ft_posting_key = function(name, term, doc_id)
	return name .. "\x03" .. term .. "\x00\x01" .. doc_id
end

local ft_reverse_key = function(name, doc_id)
	return name .. "\x04" .. doc_id
end

-- Reverse map packing: [u16 len][term_bytes]...
local ft_pack_reverse = function(term_set)
	local parts = {}
	for term in pairs(term_set) do
		local len = #term
		parts[#parts + 1] = string.char(len % 256, math.floor(len / 256))
		parts[#parts + 1] = term
	end
	return table.concat(parts)
end

local ft_unpack_reverse = function(raw)
	local terms = {}
	local pos = 1
	while pos + 1 <= #raw do
		local lo, hi = raw:byte(pos, pos + 1)
		local len = lo + hi * 256
		pos = pos + 2
		if pos + len - 1 > #raw then
			break
		end
		terms[#terms + 1] = raw:sub(pos, pos + len - 1)
		pos = pos + len
	end
	return terms
end

-- Remove all index entries for a document within an open write txn.
-- Returns old_doc_len or nil, err.
local ft_remove_doc_index = function(db, ks_name, txn, name, doc_id)
	-- Read reverse map
	local rev_raw = ks_read(txn, ks_name, ft_reverse_key(name, doc_id))
	if not rev_raw then
		return nil, "corrupt full-text index: missing reverse map"
	end

	local old_terms = ft_unpack_reverse(rev_raw)

	-- Read old doc_len
	local dm_raw = ks_read(txn, ks_name, ft_doc_meta_key(name, doc_id))
	local old_doc_len = 0
	if dm_raw and #dm_raw >= 4 then
		old_doc_len = core.unpack_u32(dm_raw)
	end

	-- Remove postings and update df for each term
	for _, term in ipairs(old_terms) do
		-- Delete posting
		local dok, derr = ks_delete(ks_name, txn, ft_posting_key(name, term, doc_id))
		if not dok then
			return nil, derr
		end

		-- Decrement df
		local df_raw = ks_read(txn, ks_name, ft_df_key(name, term))
		if df_raw and #df_raw >= 4 then
			local df = core.unpack_u32(df_raw)
			local wok, werr
			if df <= 1 then
				wok, werr = ks_delete(ks_name, txn, ft_df_key(name, term))
			else
				wok, werr = ks_write(ks_name, txn, ft_df_key(name, term), core.pack_u32(df - 1), TYPE_BYTES, 0)
			end
			if not wok then
				return nil, werr
			end
		end
	end

	-- Delete reverse map, doc text, doc metadata
	local dok, derr = ks_delete(ks_name, txn, ft_reverse_key(name, doc_id))
	if not dok then
		return nil, derr
	end
	dok, derr = ks_delete(ks_name, txn, ft_doc_key(name, doc_id))
	if not dok then
		return nil, derr
	end
	dok, derr = ks_delete(ks_name, txn, ft_doc_meta_key(name, doc_id))
	if not dok then
		return nil, derr
	end

	return old_doc_len
end

---! Index a document for full-text search
---@ ft:put(`doc_id`{.str}, `text`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Stores the document text and indexes it for BM25 search. If a document
with the same ID already exists, it is updated (old index entries are
removed first). When indexing is paused, the text is stored but not
indexed until `rebuild_index()` is called.
]===]
local ft_put = function(self, doc_id, text)
	if type(text) ~= "string" then
		return nil, "text must be a string"
	end

	local name = self.cfg.name
	local vok, verr = check_doc_id(name, doc_id)
	if not vok then
		return nil, verr
	end

	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local ks_name = self.__state.ks_name

	return with_write_txn(db, function(txn)
		-- Analyze text
		local terms
		if self.__state.analyzer then
			local ok, result = pcall(self.__state.analyzer, text)
			if not ok then
				return nil, "analyzer error: " .. tostring(result)
			end
			terms = result or {}
		else
			terms = core.ft_analyze(text)
			if not terms then
				terms = {}
			end
		end

		-- Validate custom analyzer output terms
		local tf_map = {} -- term → count
		local doc_len = #terms
		for _, term in ipairs(terms) do
			local tok, terr = check_ft_term(name, term, doc_id)
			if not tok then
				return nil, terr
			end
			tf_map[term] = (tf_map[term] or 0) + 1
		end

		local paused = ft_is_paused(self, txn)

		-- Check if document exists (update case)
		local is_update = false
		local old_doc_len = 0
		local old_doc = ks_read(txn, ks_name, ft_doc_key(name, doc_id))
		if old_doc then
			is_update = true
			if not paused then
				local rlen, rerr = ft_remove_doc_index(db, ks_name, txn, name, doc_id)
				if not rlen then
					return nil, rerr
				end
				old_doc_len = rlen
			end
		end

		-- Store document text
		local wok, werr = ks_write(ks_name, txn, ft_doc_key(name, doc_id), text, TYPE_BYTES, 0)
		if not wok then
			return nil, werr
		end

		-- Store doc metadata (doc_len as u32)
		wok, werr = ks_write(ks_name, txn, ft_doc_meta_key(name, doc_id), core.pack_u32(doc_len), TYPE_BYTES, 0)
		if not wok then
			return nil, werr
		end

		if not paused then
			-- Write index entries
			local term_set = {} -- for reverse map
			for term, tf in pairs(tf_map) do
				term_set[term] = true

				-- Update df (increment)
				local df_raw = ks_read(txn, ks_name, ft_df_key(name, term))
				local df = 0
				if df_raw and #df_raw >= 4 then
					df = core.unpack_u32(df_raw)
				end
				wok, werr = ks_write(ks_name, txn, ft_df_key(name, term), core.pack_u32(df + 1), TYPE_BYTES, 0)
				if not wok then
					return nil, werr
				end

				-- Write posting (tf)
				wok, werr = ks_write(ks_name, txn, ft_posting_key(name, term, doc_id), core.pack_u32(tf), TYPE_BYTES, 0)
				if not wok then
					return nil, werr
				end
			end

			-- Write reverse map
			wok, werr = ks_write(ks_name, txn, ft_reverse_key(name, doc_id), ft_pack_reverse(term_set), TYPE_BYTES, 0)
			if not wok then
				return nil, werr
			end

			-- Update corpus stats (only after all index writes succeed)
			local meta_raw = ks_read(txn, ks_name, ft_meta_key(name))
			local doc_count, total_doc_len = 0, 0
			if meta_raw and #meta_raw >= 12 then
				doc_count = core.unpack_u32(meta_raw:sub(1, 4))
				total_doc_len = core.unpack_int64(meta_raw:sub(5, 12))
			end

			if not is_update then
				doc_count = doc_count + 1
			else
				total_doc_len = total_doc_len - old_doc_len
				if total_doc_len < 0 then
					total_doc_len = 0
				end
			end
			total_doc_len = total_doc_len + doc_len

			local stats_val = core.pack_u32(doc_count) .. core.pack_int64(total_doc_len)
			wok, werr = ks_write(ks_name, txn, ft_meta_key(name), stats_val, TYPE_FT_META, 0)
			if not wok then
				return nil, werr
			end

			-- Write analyzer name on first write
			local existing_analyzer = ks_read(txn, ks_name, ft_analyzer_key(name))
			if not existing_analyzer then
				local aname = self.__state.analyzer_name or "default"
				wok, werr = ks_write(ks_name, txn, ft_analyzer_key(name), aname, TYPE_FT_META, 0)
				if not wok then
					return nil, werr
				end
			end
		end

		return true, true
	end)
end

---! Retrieve the original text of a document
---@ ft:get(`doc_id`{.str}) -> `text`{.str .or_nil}, `err`{.str .err}
local ft_get = function(self, doc_id)
	if type(doc_id) ~= "string" or #doc_id == 0 then
		return nil, "doc_id must be a non-empty string"
	end

	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local raw = ks_read(txn, self.__state.ks_name, ft_doc_key(self.cfg.name, doc_id))
	txn:abort()
	if not raw then
		return nil
	end
	return raw
end

---! Delete a document and its index entries
---@ ft:del(`doc_id`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local ft_del = function(self, doc_id)
	if type(doc_id) ~= "string" or #doc_id == 0 then
		return nil, "doc_id must be a non-empty string"
	end

	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local ks_name = self.__state.ks_name
	local name = self.cfg.name

	return with_write_txn(db, function(txn)
		-- Check if doc exists
		local doc_raw = ks_read(txn, ks_name, ft_doc_key(name, doc_id))
		if not doc_raw then
			return nil, "not found"
		end

		local rev_raw = ks_read(txn, ks_name, ft_reverse_key(name, doc_id))
		local had_index = rev_raw ~= nil
		local old_doc_len = 0

		if had_index then
			local rlen, rerr = ft_remove_doc_index(db, ks_name, txn, name, doc_id)
			if not rlen then
				return nil, rerr
			end
			old_doc_len = rlen
		elseif not ft_is_paused(self, txn) then
			return nil, "corrupt full-text index: missing reverse map"
		else
			local dok, derr = ks_delete(ks_name, txn, ft_doc_key(name, doc_id))
			if not dok then
				return nil, derr
			end
			dok, derr = ks_delete(ks_name, txn, ft_doc_meta_key(name, doc_id))
			if not dok then
				return nil, derr
			end
		end

		-- Update corpus stats only when index entries were removed
		if had_index then
			local meta_raw = ks_read(txn, ks_name, ft_meta_key(name))
			if meta_raw and #meta_raw >= 12 then
				local doc_count = core.unpack_u32(meta_raw:sub(1, 4))
				local total_doc_len = core.unpack_int64(meta_raw:sub(5, 12))

				doc_count = doc_count > 0 and doc_count - 1 or 0
				total_doc_len = total_doc_len > old_doc_len and total_doc_len - old_doc_len or 0

				local stats_val = core.pack_u32(doc_count) .. core.pack_int64(total_doc_len)
				local wok, werr = ks_write(ks_name, txn, ft_meta_key(name), stats_val, TYPE_FT_META, 0)
				if not wok then
					return nil, werr
				end
			end
		end
		return true, true
	end)
end

---! Check if a document exists
---@ ft:exists(`doc_id`{.str}) -> `exists`{.bool}
local ft_exists = function(self, doc_id)
	if type(doc_id) ~= "string" or #doc_id == 0 then
		return false
	end

	local db = self.__state.db
	if db.__state.closed then
		return false
	end

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return false
	end

	local raw = ks_read(txn, self.__state.ks_name, ft_doc_key(self.cfg.name, doc_id))
	txn:abort()
	return raw ~= nil
end

---! Search for documents matching a query
---@ ft:search(`query`{.str}, `opts`{.tbl .opt}) -> `results`{.tbl}
--[===[
Returns an array of `{key, score}` tables ranked by BM25 score (descending).
Options: `top_k` (max results; default 10), `k1` (BM25 k1; default 1.2),
`b` (BM25 b; default 0.75). The query is analyzed with the same analyzer
used to index the documents.
]===]
local ft_search = function(self, query_text, opts)
	if type(query_text) ~= "string" or #query_text == 0 then
		return {}
	end

	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	if self.__state.indexing_paused then
		return nil, "indexing paused"
	end

	local top_k = (opts and opts.top_k) or 10
	local k1 = (opts and opts.k1) or self.__state.k1
	local b = (opts and opts.b) or self.__state.b

	-- Analyze query
	local terms
	if self.__state.analyzer then
		local ok, result = pcall(self.__state.analyzer, query_text)
		if not ok then
			return nil, "analyzer error: " .. tostring(result)
		end
		terms = result or {}
	else
		terms = core.ft_analyze(query_text)
		if not terms then
			return {}
		end
	end

	-- Deduplicate
	local seen = {}
	local unique = {}
	for _, term in ipairs(terms) do
		if not seen[term] then
			seen[term] = true
			unique[#unique + 1] = term
		end
	end

	if #unique == 0 then
		return {}
	end

	local txn, err = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, self.__state.ks_name)
	if not ks_root then
		txn:abort()
		return {}
	end

	if ft_is_paused(self, txn) then
		txn:abort()
		return nil, "indexing paused"
	end

	local results = core.ft_search(txn, ks_root, self.cfg.name, unique, top_k, k1, b)
	txn:abort()
	return results or {}
end

---! Iterate documents by ID prefix
---@ ft:scan(`prefix`{.str .opt}) -> `iterator`{.func}
--[===[
Returns a stateful iterator that yields `(doc_id, text)` pairs for all
documents whose ID matches the prefix (or all documents if omitted).
]===]
local ft_scan = function(self, prefix)
	local db = self.__state.db
	if db.__state.closed then
		return function()
			return nil
		end
	end

	local ks_name = self.__state.ks_name
	local name = self.cfg.name

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return function()
			return nil
		end
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, ks_name)
	if not ks_root then
		txn:abort()
		return function()
			return nil
		end
	end

	local cur = txn:cursor_open(ks_root)
	if not cur then
		txn:abort()
		return function()
			return nil
		end
	end

	local doc_prefix = name .. "\x01"
	local seek_prefix = doc_prefix
	if prefix and #prefix > 0 then
		seek_prefix = doc_prefix .. prefix
	end

	local started = false
	local done = false
	local doc_prefix_len = #doc_prefix

	return function()
		while not done do
			local k, v, tag, ttl
			if not started then
				started = true
				k, v, tag, ttl = cur:seek(seek_prefix)
			else
				k, v, tag, ttl = cur:next()
			end

			if not k then
				done = true
				cur:close()
				txn:abort()
				return nil
			end

			-- Check doc_prefix
			if #k < doc_prefix_len or k:sub(1, doc_prefix_len) ~= doc_prefix then
				done = true
				cur:close()
				txn:abort()
				return nil
			end

			-- Additional prefix filter
			if prefix and #prefix > 0 then
				local doc_id = k:sub(doc_prefix_len + 1)
				if doc_id:sub(1, #prefix) ~= prefix then
					done = true
					cur:close()
					txn:abort()
					return nil
				end
				return doc_id, v
			else
				return k:sub(doc_prefix_len + 1), v
			end
		end
	end
end

---! Count documents matching a prefix
---@ ft:count(`prefix`{.str .opt}) -> `n`{.num}
local ft_count = function(self, prefix)
	local n = 0
	for _ in self:scan(prefix) do
		n = n + 1
	end
	return n
end

---! Get corpus statistics
---@ ft:stats() -> `stats`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a table with `doc_count`, `unique_terms`, `avg_doc_len`, and
`total_doc_len` fields.
]===]
local ft_stats = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end

	local ks_name = self.__state.ks_name
	local name = self.cfg.name

	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, "failed to begin transaction"
	end

	-- Read corpus stats
	local meta_raw = ks_read(txn, ks_name, ft_meta_key(name))
	local doc_count, total_doc_len = 0, 0
	if meta_raw and #meta_raw >= 12 then
		doc_count = core.unpack_u32(meta_raw:sub(1, 4))
		total_doc_len = core.unpack_int64(meta_raw:sub(5, 12))
	end

	-- Count unique terms by scanning \x03 prefix for df entries (\x00\x00 suffix)
	local unique_terms = 0
	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, ks_name)
	if ks_root then
		local idx_prefix = name .. "\x03"
		local df_suffix = "\x00\x00"
		local cur = txn:cursor_open(ks_root)
		if cur then
			local k = cur:seek(idx_prefix)
			while k do
				if #k < #idx_prefix or k:sub(1, #idx_prefix) ~= idx_prefix then
					break
				end
				-- Check if this is a df entry (ends with \x00\x00)
				if #k >= 2 and k:sub(-2) == df_suffix then
					unique_terms = unique_terms + 1
				end
				k = cur:next()
			end
			cur:close()
		end
	end

	txn:abort()

	local avg_doc_len = 0
	if doc_count > 0 then
		avg_doc_len = total_doc_len / doc_count
	end

	return {
		doc_count = doc_count,
		unique_terms = unique_terms,
		avg_doc_len = avg_doc_len,
		total_doc_len = total_doc_len,
	}
end

---! Pause automatic indexing for bulk loading
---@ ft:pause_indexing()
--[===[
After calling this, `ft:put()` stores document text but skips index
updates. Call `ft:rebuild_index()` after bulk loading to build the
index in a single batch.
]===]
local ft_pause_indexing = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end
	self.__state.indexing_paused = true
	local ks_name = self.__state.ks_name
	local name = self.cfg.name
	return with_write_txn(db, function(txn)
		local ok, werr = ks_write(ks_name, txn, ft_paused_key(name), "1", TYPE_FT_META, 0)
		if not ok then
			return nil, werr
		end
		return true, true
	end)
end

---! Rebuild the full-text index from stored documents
---@ ft:rebuild_index() -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Deletes all existing index entries and rebuilds from stored document text.
Commits in batches of 1000 documents for large corpora. Resumes automatic
indexing when complete.
]===]
local ft_rebuild_index = function(self)
	local db = self.__state.db
	if db.__state.closed then
		return nil, "db is closed"
	end
	if db.cfg.readonly then
		return nil, "db is read-only"
	end

	local ks_name = self.__state.ks_name
	local name = self.cfg.name

	-- Phase 1: Delete all existing index entries (\x03 and \x04 prefixes)
	local txn, err = db.__state.handle:txn_begin(false)
	if not txn then
		return nil, err
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = ks_dir_get(txn, dir_root, ks_name)
	if ks_root then
		-- Collect keys to delete in \x03 and \x04 regions
		for _, prefix_byte in ipairs({ "\x03", "\x04" }) do
			local pfx = name .. prefix_byte
			local to_del = {}
			local cur = txn:cursor_open(ks_root)
			if cur then
				local k = cur:seek(pfx)
				while k do
					if #k < #pfx or k:sub(1, #pfx) ~= pfx then
						break
					end
					to_del[#to_del + 1] = k
					k = cur:next()
				end
				cur:close()
			end
			for _, k in ipairs(to_del) do
				ks_delete(ks_name, txn, k)
			end
		end
	end

	-- Reset corpus stats
	ks_write(ks_name, txn, ft_meta_key(name), core.pack_u32(0) .. core.pack_int64(0), TYPE_FT_META, 0)

	local ok, cerr = txn:commit()
	if not ok then
		return nil, cerr
	end

	-- Phase 2: Rebuild from stored documents
	local doc_count = 0
	local total_doc_len = 0
	local batch_count = 0
	local batch_limit = 1000

	txn, err = db.__state.handle:txn_begin(false)
	if not txn then
		return nil, err
	end

	-- Scan all documents
	local doc_prefix = name .. "\x01"
	dir_root = txn:meta().keyspace_root_pgno
	ks_root = ks_dir_get(txn, dir_root, ks_name)

	if ks_root then
		-- Collect doc ids first (a cursor would be invalidated by the index
		-- writes); text is re-read per document below so a multi-GB corpus
		-- is never held in memory wholesale.
		local docs = {}
		local read_txn = db.__state.handle:txn_begin(true)
		if read_txn then
			local rd_root = ks_dir_get(read_txn, read_txn:meta().keyspace_root_pgno, ks_name)
			if rd_root then
				local cur = read_txn:cursor_open(rd_root)
				if cur then
					local k = cur:seek(doc_prefix)
					while k do
						if #k < #doc_prefix or k:sub(1, #doc_prefix) ~= doc_prefix then
							break
						end
						docs[#docs + 1] = k:sub(#doc_prefix + 1)
						k = cur:next()
					end
					cur:close()
				end
			end
			read_txn:abort()
		end

		for _, doc_id in ipairs(docs) do
			local doc = { id = doc_id, text = ks_read(txn, ks_name, ft_doc_key(name, doc_id)) }
			if not doc.text then
				txn:abort()
				self.__state.indexing_paused = true
				return nil, "document disappeared during rebuild"
			end
			-- Analyze text
			local terms
			if self.__state.analyzer then
				local aok, result = pcall(self.__state.analyzer, doc.text)
				if not aok then
					txn:abort()
					self.__state.indexing_paused = true -- keep paused; rebuild failed
					return nil, "analyzer error: " .. tostring(result)
				end
				terms = result or {}
			else
				terms = core.ft_analyze(doc.text) or {}
			end

			-- Count term frequencies and validate terms
			local tf_map = {}
			local doc_len = #terms
			for _, term in ipairs(terms) do
				local tok, terr = check_ft_term(name, term, doc.id)
				if not tok then
					txn:abort()
					self.__state.indexing_paused = true -- keep paused; rebuild failed
					return nil, terr
				end
				tf_map[term] = (tf_map[term] or 0) + 1
			end

			-- Write index entries
			local term_set = {}
			for term, tf in pairs(tf_map) do
				term_set[term] = true

				-- Update df
				local df_raw = ks_read(txn, ks_name, ft_df_key(name, term))
				local df = 0
				if df_raw and #df_raw >= 4 then
					df = core.unpack_u32(df_raw)
				end
				local wok, werr = ks_write(ks_name, txn, ft_df_key(name, term), core.pack_u32(df + 1), TYPE_BYTES, 0)
				if not wok then
					txn:abort()
					self.__state.indexing_paused = true
					return nil, werr
				end

				-- Write posting
				wok, werr = ks_write(ks_name, txn, ft_posting_key(name, term, doc.id), core.pack_u32(tf), TYPE_BYTES, 0)
				if not wok then
					txn:abort()
					self.__state.indexing_paused = true
					return nil, werr
				end
			end

			-- Write reverse map
			local wok, werr =
				ks_write(ks_name, txn, ft_reverse_key(name, doc.id), ft_pack_reverse(term_set), TYPE_BYTES, 0)
			if not wok then
				txn:abort()
				self.__state.indexing_paused = true
				return nil, werr
			end

			-- Update doc metadata (doc_len) in case it was missing during paused put
			wok, werr = ks_write(ks_name, txn, ft_doc_meta_key(name, doc.id), core.pack_u32(doc_len), TYPE_BYTES, 0)
			if not wok then
				txn:abort()
				self.__state.indexing_paused = true
				return nil, werr
			end

			doc_count = doc_count + 1
			total_doc_len = total_doc_len + doc_len
			batch_count = batch_count + 1

			-- Periodic commit for large corpora
			if batch_count >= batch_limit then
				-- Write partial corpus stats
				wok, werr = ks_write(
					ks_name,
					txn,
					ft_meta_key(name),
					core.pack_u32(doc_count) .. core.pack_int64(total_doc_len),
					TYPE_FT_META,
					0
				)
				if not wok then
					txn:abort()
					self.__state.indexing_paused = true
					return nil, werr
				end

				ok, cerr = txn:commit()
				if not ok then
					self.__state.indexing_paused = true
					return nil, cerr
				end

				txn, err = db.__state.handle:txn_begin(false)
				if not txn then
					self.__state.indexing_paused = true
					return nil, err
				end
				batch_count = 0
			end
		end
	end

	-- Final corpus stats and commit
	local wok, werr = ks_write(
		ks_name,
		txn,
		ft_meta_key(name),
		core.pack_u32(doc_count) .. core.pack_int64(total_doc_len),
		TYPE_FT_META,
		0
	)
	if not wok then
		txn:abort()
		self.__state.indexing_paused = true
		return nil, werr
	end

	-- Clear the persisted paused marker atomically with the final batch:
	-- a crash before this commit leaves the index flagged incomplete.
	if ks_read(txn, ks_name, ft_paused_key(name)) then
		local dok, derr = ks_delete(ks_name, txn, ft_paused_key(name))
		if not dok then
			txn:abort()
			self.__state.indexing_paused = true
			return nil, derr
		end
	end

	ok, cerr = txn:commit()
	if not ok then
		self.__state.indexing_paused = true
		return nil, cerr
	end

	self.__state.indexing_paused = false
	return true
end

-- FT keyspace constructor
local new_ft_keyspace = function(db, name, opts)
	if opts and opts.encrypted then
		return nil, "encryption not supported for full-text keyspaces"
	end
	local analyzer = opts and opts.analyzer
	local analyzer_name = opts and opts.analyzer_name
	local k1 = (opts and opts.k1) or 1.2
	local b_val = (opts and opts.b) or 0.75

	-- The FT keyspace uses a dedicated MNEME keyspace.
	-- The ks_name is the FT name itself (it gets its own B-tree).
	local ks_name = name

	-- Check analyzer consistency and pause state if keyspace already exists
	local persisted_paused = false
	if not db.__state.closed then
		local txn = db.__state.handle:txn_begin(true)
		if txn then
			persisted_paused = ks_read(txn, ks_name, ft_paused_key(name)) ~= nil
			local existing = ks_read(txn, ks_name, ft_analyzer_key(name))
			if existing then
				local expected = analyzer_name or "default"
				if existing ~= expected then
					io.stderr:write(
						"mneme: warning: FT keyspace '"
							.. name
							.. "' was built with analyzer '"
							.. existing
							.. "' but opened with '"
							.. expected
							.. "'\n"
					)
				end
			end
			txn:abort()
		end
	end

	return {
		cfg = { name = name },
		__state = {
			db = db,
			ks_name = ks_name,
			analyzer = analyzer,
			analyzer_name = analyzer_name,
			k1 = k1,
			b = b_val,
			indexing_paused = persisted_paused,
		},
		put = ft_put,
		get = ft_get,
		del = ft_del,
		exists = ft_exists,
		search = ft_search,
		scan = ft_scan,
		count = ft_count,
		stats = ft_stats,
		pause_indexing = ft_pause_indexing,
		rebuild_index = ft_rebuild_index,
	}
end

---! Get or create a full-text search keyspace
---@ db:ft_keyspace(`name`{.str}, `opts`{.tbl .opt}) -> `ft`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a full-text keyspace handle with BM25 ranked search. Options:
`analyzer` (custom tokenizer function), `analyzer_name` (string label),
`k1` (BM25 k1; default 1.2), `b` (BM25 b; default 0.75). Encryption
is not supported for FT keyspaces.
]===]
local db_ft_keyspace = function(self, name, opts)
	if self.__state.closed then
		return nil, "db is closed"
	end
	local ok, cerr = check_keyspace_name(name)
	if not ok then
		return nil, cerr
	end
	return new_ft_keyspace(self, name, opts)
end

-- ── Keyspace constructor ──────────────────────────────────────────────

-- Persisted per-keyspace encryption marker, stored in the keyspace
-- directory next to the root entries ("1" = encrypted, "0" = plaintext).
-- Without it the encrypted status would exist only as a per-handle option:
-- reopening "secrets" without `encrypted = true` would read ciphertext as
-- values and write plaintext into it, silently.
local ks_enc_marker_key = function(name)
	return "\x00ksenc:" .. name
end

local new_keyspace = function(db, name, opts)
	local want_encrypted = (opts and opts.encrypted) and true or false
	if want_encrypted and not db.__state.dek then
		return nil, "encryption requested but database has no encryption key"
	end

	-- Compare against the persisted marker; fail on mismatch in either
	-- direction. Keyspaces created before markers existed have none and
	-- get one codifying this first claim (best effort, skipped when
	-- read-only or another write txn is active).
	local txn = db.__state.handle:txn_begin(true)
	if not txn then
		return nil, "failed to begin transaction"
	end
	local marker = txn:btree_get(txn:meta().keyspace_root_pgno, ks_enc_marker_key(name))
	txn:abort()

	if marker then
		local marked_encrypted = (marker == "1")
		if marked_encrypted and not want_encrypted then
			return nil, "keyspace is encrypted; pass encrypted = true"
		end
		if not marked_encrypted and want_encrypted then
			return nil, "keyspace is not encrypted; opening it encrypted would mix plaintext and ciphertext"
		end
	elseif not db.cfg.readonly then
		local wtxn = db.__state.handle:txn_begin(false)
		if wtxn then
			local wdir = wtxn:meta().keyspace_root_pgno
			local nd = wtxn:btree_put(wdir, ks_enc_marker_key(name), want_encrypted and "1" or "0", TYPE_BYTES, 0)
			if nd then
				if nd ~= wdir then
					wtxn:set_ks_root(nd)
				end
				wtxn:commit()
			else
				wtxn:abort()
			end
		end
	end

	-- Derive per-keyspace sub-key if encryption is active
	local subkey = nil
	if want_encrypted then
		subkey = crypto.derive_subkey(db.__state.dek, name)
	end
	return {
		cfg = { name = name, encrypted = (subkey ~= nil) },
		__state = { db = db, subkey = subkey },
		get = ks_get,
		set = ks_set,
		set_bytes = ks_set_bytes,
		set_integer = ks_set_integer,
		set_float = ks_set_float,
		del = ks_del,
		exists = ks_exists,
		incr = ks_incr,
		decr = ks_decr,
		ttl = ks_ttl,
		persist = ks_persist,
		purge_expired = ks_purge_expired,
		scan = ks_scan,
		range = ks_range,
		count = ks_count,
		batch = ks_batch,
		stats = ks_stats,
		sorted_set = ks_sorted_set,
		list = ks_list,
		vset = ks_vset,
		vget = ks_vget,
		vsearch = ks_vsearch,
		vdim = ks_vdim,
	}
end

-- ── Database methods ──────────────────────────────────────────────────

---! Close the database
---@ db:close() -> `ok`{.bool .or_nil}, `err`{.str .err}
local db_close = function(self)
	if self.__state.closed then
		return nil, "db is closed"
	end
	local ok, err = self.__state.handle:close()
	if not ok then
		-- e.g. a write transaction is still active; closing now would
		-- free the handle out from under it.
		return nil, err
	end
	self.__state.closed = true
	return true
end

---! Flush the database file to stable storage (fdatasync)
---@ db:sync() -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Explicitly fdatasyncs the database file. The complement of `sync = "none"`:
a store opened without per-commit syncs calls this at its own durability
points (a bulk load's end, an application's safe checkpoints).
]===]
local db_sync = function(self)
	if self.__state.closed then
		return nil, "db is closed"
	end
	local ok, err = self.__state.handle:sync()
	if not ok then
		return nil, err
	end
	return true
end

---! Get low-level database info
---@ db:info() -> `info`{.tbl .or_nil}, `err`{.str .err}
local db_info = function(self)
	if self.__state.closed then
		return nil, "db is closed"
	end
	local info, err = self.__state.handle:info()
	if not info then
		return nil, err
	end
	return info
end

local db_txn_begin = function(self, readonly)
	if self.__state.closed then
		return nil, "db is closed"
	end
	return self.__state.handle:txn_begin(readonly)
end

---! Get or create a keyspace
---@ db:keyspace(`name`{.str}, `opts`{.tbl .opt}) -> `ks`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a keyspace handle. The keyspace is created on first write if it
does not exist. Pass `opts.encrypted = true` to enable per-keyspace
encryption (requires the database to be opened with an encryption key).
]===]
local db_keyspace = function(self, name, opts)
	if self.__state.closed then
		return nil, "db is closed"
	end
	local ok, cerr = check_keyspace_name(name)
	if not ok then
		return nil, cerr
	end
	return new_keyspace(self, name, opts)
end

---! List all keyspace names
---@ db:keyspaces() -> `names`{.tbl .or_nil}, `err`{.str .err}
local db_keyspaces = function(self)
	if self.__state.closed then
		return nil, "db is closed"
	end

	local txn = self.__state.handle:txn_begin(true)
	if not txn then
		return nil, "failed to begin transaction"
	end

	local dir_root = txn:meta().keyspace_root_pgno
	local cur = txn:cursor_open(dir_root)
	if not cur then
		txn:abort()
		return {}
	end

	local names = {}
	local k = cur:first()
	while k do
		-- Filter out internal encryption metadata key
		if k:sub(1, 1) ~= "\x00" then
			table.insert(names, k)
		end
		k = cur:next()
	end
	cur:close()
	txn:abort()
	return names
end

---! Drop a keyspace and its data
---@ db:drop_keyspace(`name`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
local db_drop_keyspace = function(self, name)
	if self.__state.closed then
		return nil, "db is closed"
	end
	if self.cfg.readonly then
		return nil, "db is read-only"
	end
	local ok, cerr = check_keyspace_name(name)
	if not ok then
		return nil, cerr
	end

	return with_write_txn(self, function(txn)
		local dir_root = txn:meta().keyspace_root_pgno

		-- Record the keyspace's whole subtree as reclaimable by compaction.
		-- Current COW reclamation does not link freed pages into the persistent
		-- free list yet; db:compact() rewrites live data to reclaim them.
		local ks_root = ks_dir_get(txn, dir_root, name)
		if ks_root and ks_root ~= 0 then
			local fok, ferr = txn:btree_free(ks_root)
			if not fok then
				return nil, ferr or "btree free failed"
			end
		end

		local new_dir_root = ks_dir_del(txn, dir_root, name)
		if new_dir_root and new_dir_root ~= dir_root then
			txn:set_ks_root(new_dir_root)
		end

		-- Drop the encryption marker too, so the name can be reused
		-- in either mode.
		dir_root = txn:meta().keyspace_root_pgno
		new_dir_root = ks_dir_del(txn, dir_root, ks_enc_marker_key(name))
		if new_dir_root and new_dir_root ~= dir_root then
			txn:set_ks_root(new_dir_root)
		end

		return true, true
	end)
end

-- ── Encryption key management ─────────────────────────────────────────

-- Read current key slots from the database.
-- Returns slots array or nil (no encryption).
local enc_read_slots = function(self)
	local txn = self.__state.handle:txn_begin(true)
	if not txn then
		return nil, "failed to begin transaction"
	end
	local dir_root = txn:meta().keyspace_root_pgno
	local enc_raw = txn:btree_get(dir_root, crypto.ENC_META_KEY)
	txn:abort()
	if not enc_raw then
		return nil
	end
	return crypto.unpack_slots(enc_raw)
end

-- Write key slots to the database.
local enc_write_slots = function(self, slots)
	local slots_blob = crypto.pack_slots(slots)

	local txn = self.__state.handle:txn_begin(false)
	if not txn then
		return nil, "failed to begin transaction"
	end
	local dir_root = txn:meta().keyspace_root_pgno
	local new_dir = txn:btree_put(dir_root, crypto.ENC_META_KEY, slots_blob, TYPE_BYTES, 0)
	if not new_dir then
		txn:abort()
		return nil, "failed to write encryption metadata"
	end
	if new_dir ~= dir_root then
		txn:set_ks_root(new_dir)
	end
	return txn:commit()
end

---! Authorize an additional encryption key
---@ db:add_encryption_key(`pubkey`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Adds a new Ed25519 public key (32 bytes) to the database's authorized key
list. The current key (from `encryption.key_file`) is used to unwrap the
DEK, which is then re-wrapped for the new key.
]===]
local db_add_encryption_key = function(self, new_ed25519_pubkey)
	if self.__state.closed then
		return nil, "db is closed"
	end
	if self.cfg.readonly then
		return nil, "db is read-only"
	end
	if not self.__state.dek then
		return nil, "database encryption not enabled"
	end

	-- Re-load our own key to unwrap the DEK
	local own_seed, own_pubkey
	if self.__state.enc_key_file then
		own_seed, own_pubkey = crypto.load_ssh_key(self.__state.enc_key_file)
		if not own_seed then
			return nil, "failed to re-read key file: " .. (own_pubkey or "unknown")
		end
	else
		return nil, "key management requires key_file in encryption options"
	end

	-- Read current slots
	local slots, rerr = enc_read_slots(self)
	if not slots then
		return nil, rerr or "no encryption metadata"
	end

	-- Check for duplicate
	local fp = crypto.fingerprint(new_ed25519_pubkey)
	for _, slot in ipairs(slots) do
		if slot.fingerprint == fp then
			return nil, "key already authorized"
		end
	end

	-- Unwrap DEK using our key, get raw bytes for re-wrapping
	local dek_raw, uerr = crypto.unwrap_dek_raw(slots, own_seed, own_pubkey)
	if not dek_raw then
		return nil, uerr
	end

	-- Wrap for the new key (only needs their public key)
	local new_slot = crypto.wrap_dek(dek_raw, new_ed25519_pubkey)
	table.insert(slots, new_slot)

	return enc_write_slots(self, slots)
end

---! Revoke an authorized encryption key
---@ db:remove_encryption_key(`fingerprint`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Removes the key slot matching the given 32-byte SHA-256 fingerprint.
Cannot remove the last remaining key slot.
]===]
local db_remove_encryption_key = function(self, fingerprint)
	if self.__state.closed then
		return nil, "db is closed"
	end
	if self.cfg.readonly then
		return nil, "db is read-only"
	end
	if not self.__state.dek then
		return nil, "database encryption not enabled"
	end

	local slots, rerr = enc_read_slots(self)
	if not slots then
		return nil, rerr or "no encryption metadata"
	end

	if #slots <= 1 then
		return nil, "cannot remove last key slot"
	end

	local new_slots = {}
	local found = false
	for _, slot in ipairs(slots) do
		if slot.fingerprint == fingerprint then
			found = true
		else
			new_slots[#new_slots + 1] = slot
		end
	end

	if not found then
		return nil, "key not found"
	end

	return enc_write_slots(self, new_slots)
end

---! List authorized encryption keys
---@ db:list_encryption_keys() -> `keys`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns an array of `{fingerprint}` tables, one per authorized key slot.
]===]
local db_list_encryption_keys = function(self)
	if self.__state.closed then
		return nil, "db is closed"
	end
	if not self.__state.dek then
		return nil, "database encryption not enabled"
	end

	local slots, rerr = enc_read_slots(self)
	if not slots then
		return nil, rerr or "no encryption metadata"
	end

	local keys = {}
	for _, slot in ipairs(slots) do
		keys[#keys + 1] = { fingerprint = slot.fingerprint }
	end
	return keys
end

-- ── Database statistics ────────────────────────────────────────────────

---! Get database-level statistics
---@ db:stats() -> `stats`{.tbl .or_nil}, `err`{.str .err}
--[===[
Returns a table with `page_count`, `retired_pages`, `live_pages`,
`reclaimable_pages`, `bloat_ratio`, `keyspaces`, and `file_size` fields.
]===]
local db_stats = function(self)
	if self.__state.closed then
		return nil, "db is closed"
	end

	local txn, err = self.__state.handle:txn_begin(true)
	if not txn then
		return nil, err
	end

	local meta = txn:meta()
	local retired_pages = meta.retired_page_count or 0
	local live_pages = txn:live_count()
	local reclaimable_pages = meta.page_count - live_pages
	if reclaimable_pages < 0 then
		reclaimable_pages = 0
	end
	local bloat_ratio = 0
	if live_pages > 0 then
		bloat_ratio = reclaimable_pages / live_pages
	end

	local dir_root = meta.keyspace_root_pgno
	local ks_count = 0
	local cur = txn:cursor_open(dir_root)
	if cur then
		local k = cur:first()
		while k do
			if k:sub(1, 1) ~= "\x00" then
				ks_count = ks_count + 1
			end
			k = cur:next()
		end
		cur:close()
	end

	txn:abort()

	return {
		page_count = meta.page_count,
		retired_pages = retired_pages,
		live_pages = live_pages,
		reclaimable_pages = reclaimable_pages,
		bloat_ratio = bloat_ratio,
		keyspaces = ks_count,
		file_size = meta.page_count * PAGE_SIZE,
	}
end

-- ── Database compaction ───────────────────────────────────────────────

-- Dirty page lookup is a linear scan (O(n) per access), so a single
-- write transaction that inserts hundreds of thousands of keys becomes
-- O(n²).  We commit in batches to keep the dirty array small.
local COMPACT_BATCH = 10000

---! Compact the database to a new file
---@ db:compact(`output_path`{.str}) -> `ok`{.bool .or_nil}, `err`{.str .err}
--[===[
Copies all live (non-expired) data to a new database file, reclaiming
free pages and removing fragmentation. The current database is not
modified. Commits in batches of 10000 keys per keyspace.
]===]
local db_compact = function(self, output_path)
	if self.__state.closed then
		return nil, "db is closed"
	end
	if type(output_path) ~= "string" then
		return nil, "output path must be a string"
	end
	if output_path == self.cfg.path then
		return nil, "output path must differ from source path"
	end
	-- Refuse to clobber an existing file: compact must produce a fresh DB.
	local probe = io.open(output_path, "rb")
	if probe then
		probe:close()
		return nil, "output path exists"
	end

	-- Open source read txn
	local src_txn, err = self.__state.handle:txn_begin(true)
	if not src_txn then
		return nil, err
	end

	-- Open destination database
	local dst_db, derr = core.db_open(output_path, { sync = "none" })
	if not dst_db then
		src_txn:abort()
		return nil, derr or "failed to create output database"
	end

	local now = os.time()
	local src_dir_root = src_txn:meta().keyspace_root_pgno

	-- Iterate all keyspaces in directory
	local dir_cur = src_txn:cursor_open(src_dir_root)
	if not dir_cur then
		src_txn:abort()
		dst_db:close()
		return nil, "failed to open directory cursor"
	end

	-- Helper: flush current keyspace root into the destination directory
	-- and commit the transaction.  Returns new_txn, new_ks_root or nil, err.
	local flush_batch = function(dst_txn, ks_name, dst_ks_root)
		-- Write current root into directory
		local dst_dir_root = dst_txn:meta().keyspace_root_pgno
		local new_dir = dst_txn:btree_put(dst_dir_root, ks_name, core.pack_u32(dst_ks_root), TYPE_BYTES, 0)
		if new_dir and new_dir ~= dst_dir_root then
			dst_txn:set_ks_root(new_dir)
		end
		local ok, cerr = dst_txn:commit()
		if not ok then
			return nil, cerr or "batch commit failed"
		end

		-- Start fresh transaction, read back the persisted root
		local new_txn = dst_db:txn_begin(false)
		if not new_txn then
			return nil, "failed to begin batch txn"
		end
		local dir_root = new_txn:meta().keyspace_root_pgno
		local dir_cur2 = new_txn:cursor_open(dir_root)
		if not dir_cur2 then
			new_txn:abort()
			return nil, "failed to open directory cursor"
		end
		local found_name, new_root_raw = dir_cur2:seek(ks_name)
		dir_cur2:close()
		if not found_name or found_name ~= ks_name then
			new_txn:abort()
			return nil, "lost keyspace root after batch commit"
		end
		return new_txn, core.unpack_u32(new_root_raw)
	end

	local compact_err
	local ks_name, ks_root_raw = dir_cur:first()
	while ks_name do
		if ks_name:sub(1, 1) == "\x00" then
			-- Internal metadata (e.g. encryption key slots): copy raw value verbatim
			local dst_txn = dst_db:txn_begin(false)
			if not dst_txn then
				compact_err = "failed to begin destination txn for metadata"
				break
			end
			local dst_dir_root = dst_txn:meta().keyspace_root_pgno
			local new_dir = dst_txn:btree_put(dst_dir_root, ks_name, ks_root_raw, TYPE_BYTES, 0)
			if not new_dir then
				dst_txn:abort()
				compact_err = "failed to write metadata during compaction"
				break
			end
			if new_dir ~= dst_dir_root then
				dst_txn:set_ks_root(new_dir)
			end
			local ok, cerr = dst_txn:commit()
			if not ok then
				compact_err = cerr or "metadata commit failed"
				break
			end
		else
			local ks_root = core.unpack_u32(ks_root_raw)

			-- Begin write txn on destination for this keyspace
			local dst_txn = dst_db:txn_begin(false)
			if not dst_txn then
				compact_err = "failed to begin destination txn"
				break
			end

			-- Allocate empty leaf for keyspace in destination
			local dst_leaf = dst_txn:alloc_leaf()
			if not dst_leaf then
				dst_txn:abort()
				compact_err = "failed to allocate leaf page"
				break
			end
			local dst_ks_root = dst_leaf

			-- Iterate all keys in the source keyspace
			local ks_ok = true
			local ks_cur = src_txn:cursor_open(ks_root)
			if ks_cur then
				local batch_count = 0
				local k, v, tag, ttl = ks_cur:first()
				while k do
					-- Skip expired keys
					if ttl == 0 or ttl >= now then
						dst_ks_root = dst_txn:btree_put(dst_ks_root, k, v, tag, ttl)
						if not dst_ks_root then
							compact_err = "btree_put failed during compaction"
							ks_ok = false
							break
						end
						batch_count = batch_count + 1

						if batch_count >= COMPACT_BATCH then
							dst_txn, dst_ks_root = flush_batch(dst_txn, ks_name, dst_ks_root)
							if not dst_txn then
								compact_err = dst_ks_root -- flush_batch returns nil, err
								ks_ok = false
								break
							end
							batch_count = 0
						end
					end
					k, v, tag, ttl = ks_cur:next()
				end
				ks_cur:close()
			end

			if not ks_ok then
				-- dst_txn is nil when flush_batch failed (it commits or
				-- aborts internally); on a put failure it is still live
				-- and must be aborted before dst_db:close() below.
				if dst_txn then
					dst_txn:abort()
				end
				break
			end

			-- Final commit for this keyspace (registers root in directory)
			local dst_dir_root = dst_txn:meta().keyspace_root_pgno
			local new_dir = dst_txn:btree_put(dst_dir_root, ks_name, core.pack_u32(dst_ks_root), TYPE_BYTES, 0)
			if new_dir and new_dir ~= dst_dir_root then
				dst_txn:set_ks_root(new_dir)
			end
			local ok, cerr = dst_txn:commit()
			if not ok then
				compact_err = cerr or "final keyspace commit failed"
				break
			end
		end

		ks_name, ks_root_raw = dir_cur:next()
	end
	dir_cur:close()
	src_txn:abort()
	if not compact_err then
		-- The destination was opened with sync = "none"; make the result
		-- durable before reporting success.
		if not dst_db:sync() then
			compact_err = "failed to sync output database"
		end
	end
	dst_db:close()

	if compact_err then
		-- Leave no partial output behind: a retry would otherwise fail
		-- on the "output path exists" check above.
		os.remove(output_path)
		return nil, compact_err
	end
	return true
end

-- ── Module ────────────────────────────────────────────────────────────

---! Open or create a MNEME database
---@ open(`path`{.str}, `opts`{.tbl .opt}) -> `db`{.tbl .or_nil}, `err`{.str .err}
--[===[
Opens the database file at path. Creates it if it does not exist (unless
`opts.create` is false). Options: `readonly` (open read-only),
`create` (set to false to fail if file does not exist),
`sync` ("none" to disable fsync, useful for bulk loading),
`encryption` (table with `key_file` path or `seed`+`pubkey` for
Ed25519-based encryption at rest).
]===]
local open = function(path, opts)
	if type(path) ~= "string" then
		return nil, "path must be a string"
	end

	local flags = {}
	if opts then
		if opts.readonly then
			flags.readonly = true
		end
		if opts.create == false then
			flags.create = false
		end
		if opts.sync then
			flags.sync = opts.sync
		end
	end

	local handle, err = core.db_open(path, flags)
	if not handle then
		return nil, err
	end

	-- Always probe for existing encryption metadata so an encrypted DB cannot
	-- be opened as plaintext.
	local probe_enc_raw
	do
		local probe_txn, probe_err = handle:txn_begin(true)
		if not probe_txn then
			handle:close()
			return nil, probe_err or "failed to begin transaction"
		end
		local dir_root = probe_txn:meta().keyspace_root_pgno
		probe_enc_raw = probe_txn:btree_get(dir_root, crypto.ENC_META_KEY)
		probe_txn:abort()
	end
	if probe_enc_raw and not (opts and opts.encryption) then
		handle:close()
		return nil, "database is encrypted; encryption option required"
	end

	-- Handle encryption setup
	local dek = nil
	if opts and opts.encryption then
		local enc = opts.encryption

		-- Load the user's Ed25519 key
		local seed, pubkey
		if enc.key_file then
			seed, pubkey = crypto.load_ssh_key(enc.key_file)
			if not seed then
				handle:close()
				return nil, "failed to load SSH key: " .. (pubkey or "unknown error")
			end
		elseif enc.seed and enc.pubkey then
			seed = enc.seed
			pubkey = enc.pubkey
		else
			handle:close()
			return nil, "encryption requires key_file or seed+pubkey"
		end

		local enc_raw = probe_enc_raw

		if enc_raw then
			-- Database already has encryption: unwrap DEK
			local slots, uerr = crypto.unpack_slots(enc_raw)
			if not slots then
				handle:close()
				return nil, uerr
			end
			dek, err = crypto.unwrap_dek(slots, seed, pubkey)
			if not dek then
				handle:close()
				return nil, err
			end
		else
			-- No encryption metadata yet: initialize encryption
			if flags.readonly then
				handle:close()
				return nil, "cannot initialize encryption on read-only database"
			end

			-- Refuse to silently mint a fresh DEK for a database that already
			-- holds data: if its encryption metadata was lost or stripped, a
			-- new DEK would permanently orphan every encrypted value.
			if not enc.init then
				local has_data = false
				local ptxn = handle:txn_begin(true)
				if ptxn then
					local pcur = ptxn:cursor_open(ptxn:meta().keyspace_root_pgno)
					if pcur then
						local k = pcur:first()
						while k do
							if k:sub(1, 1) ~= "\x00" then
								has_data = true
								break
							end
							k = pcur:next()
						end
						pcur:close()
					end
					ptxn:abort()
				end
				if has_data then
					handle:close()
					return nil, "database already contains data; pass encryption.init = true to initialize encryption"
				end
			end

			-- Generate DEK and create first key slot
			local dek_raw = litls.random_bytes(32)
			local slot = crypto.wrap_dek(dek_raw, pubkey)
			local slots_blob = crypto.pack_slots({ slot })

			-- Write encryption metadata to keyspace directory
			local wtxn = handle:txn_begin(false)
			if not wtxn then
				handle:close()
				return nil, "failed to begin write transaction"
			end

			local dir_root = wtxn:meta().keyspace_root_pgno
			local new_dir = wtxn:btree_put(dir_root, crypto.ENC_META_KEY, slots_blob, TYPE_BYTES, 0)
			if not new_dir then
				wtxn:abort()
				handle:close()
				return nil, "failed to write encryption metadata"
			end
			if new_dir ~= dir_root then
				wtxn:set_ks_root(new_dir)
			end
			local cok, cerr = wtxn:commit()
			if not cok then
				handle:close()
				return nil, "failed to commit encryption metadata: " .. (cerr or "unknown")
			end

			dek = core.dek_new(dek_raw)
		end
	end

	local db = {
		cfg = {
			path = path,
			readonly = opts and opts.readonly or false,
			encrypted = (dek ~= nil),
		},
		__state = {
			handle = handle,
			closed = false,
			dek = dek,
			enc_key_file = opts and opts.encryption and opts.encryption.key_file,
		},
		close = db_close,
		sync = db_sync,
		info = db_info,
		stats = db_stats,
		compact = db_compact,
		txn_begin = db_txn_begin,
		keyspace = db_keyspace,
		ft_keyspace = db_ft_keyspace,
		keyspaces = db_keyspaces,
		drop_keyspace = db_drop_keyspace,
		add_encryption_key = db_add_encryption_key,
		remove_encryption_key = db_remove_encryption_key,
		list_encryption_keys = db_list_encryption_keys,
	}
	return db
end

return { open = open }