-- SPDX-FileCopyrightText: © 2026 Vladimir Zorin <vladimir@deviant.guru>
-- SPDX-License-Identifier: OWL-1.0 or later

-- Regression tests for the 2026-06 review fixes: transaction safety
-- guards, key/name validation, sorted-set semantics, FT pause
-- persistence, and per-keyspace encryption markers.

local testimony = require("testimony")
local std = require("std")
local litls = require("litls.core")
local mneme = require("mneme")
local core = require("mneme.core")

local testify = testimony.new("== mneme: review regressions ==")

local tmp_path = function()
	-- std.nanoid() is deterministic per process, so paths repeat across
	-- runs; clear any leftover so encryption-init guards see a fresh file.
	local p = "/tmp/mneme_review_" .. std.nanoid() .. ".mneme"
	std.fs.remove(p)
	return p
end

-- ── Transaction safety guards ─────────────────────────────────────────

testify:that("write methods on a read-only txn fail cleanly", function()
	local path = tmp_path()
	local db = core.db_open(path)
	local txn = db:txn_begin(true)

	local ok, err = pcall(function()
		return txn:btree_put(2, "k", "v")
	end)
	testimony.assert_true(not ok)
	testimony.assert_true(tostring(err):find("read%-only") ~= nil)

	txn:abort()
	db:close()
	std.fs.remove(path)
end)

testify:that("nested write transactions are rejected", function()
	local path = tmp_path()
	local db = core.db_open(path)

	local w = db:txn_begin(false)
	testimony.assert_not_nil(w)
	local w2, err = db:txn_begin(false)
	testimony.assert_nil(w2)
	testimony.assert_equal("write transaction already active", err)

	w:abort()
	-- After abort a new write txn must succeed again
	local w3 = db:txn_begin(false)
	testimony.assert_not_nil(w3)
	w3:abort()
	db:close()
	std.fs.remove(path)
end)

testify:that("close with an active write transaction is refused", function()
	local path = tmp_path()
	local db = core.db_open(path)

	local w = db:txn_begin(false)
	local ok, err = db:close()
	testimony.assert_nil(ok)
	testimony.assert_equal("write transaction active", err)

	w:abort()
	testimony.assert_true(db:close() == true)
	std.fs.remove(path)
end)

testify:that("batch object used after commit errors instead of crashing", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("t")

	local stashed
	local ok = ks:batch(function(b)
		b:set("k", "v")
		stashed = b
	end)
	testimony.assert_true(ok == true)

	local late_ok, late_err = pcall(function()
		stashed:set("k2", "v2")
	end)
	testimony.assert_true(not late_ok)
	testimony.assert_true(tostring(late_err):find("finished") ~= nil)
	-- The late write must not have landed
	testimony.assert_nil(ks:get("k2"))

	db:close()
	std.fs.remove(path)
end)

testify:that("cursor used after its txn finished errors instead of crashing", function()
	local path = tmp_path()
	local db = core.db_open(path)
	local txn = db:txn_begin(true)
	local cur = txn:cursor_open(2)
	txn:abort()

	local ok, err = pcall(function()
		return cur:first()
	end)
	testimony.assert_true(not ok)
	testimony.assert_true(tostring(err):find("finished") ~= nil)

	cur:close()
	db:close()
	std.fs.remove(path)
end)

testify:that("btree_put rejects forged type tags", function()
	local path = tmp_path()
	local db = core.db_open(path)
	local txn = db:txn_begin(false)

	-- 0x80 would forge the overflow flag and make readers chase an
	-- arbitrary page number
	local ok, err = txn:btree_put(2, "k", "AAAA", 0x81)
	testimony.assert_nil(ok)
	testimony.assert_equal("invalid type_tag", err)

	txn:abort()
	db:close()
	std.fs.remove(path)
end)

-- ── Key and name validation ───────────────────────────────────────────

testify:that("plain keys must not contain NUL bytes", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("t")

	-- This key would alias a phantom member of sorted set "board"
	local ok, err = ks:set("board\x00\x01junk", "x")
	testimony.assert_nil(ok)
	testimony.assert_equal("key must not contain null bytes", err)

	local ok2, err2 = ks:set("", "x")
	testimony.assert_nil(ok2)
	testimony.assert_equal("key must not be empty", err2)

	-- The vector dimension record moved under the reserved \x00 prefix,
	-- so "__mneme_vdim" is now an ordinary, usable key.
	testimony.assert_true(ks:set("__mneme_vdim", "x") == true)
	testimony.assert_equal("x", ks:get("__mneme_vdim"))

	db:close()
	std.fs.remove(path)
end)

testify:that("the reserved vector-dimension record is invisible to scan/count", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("v")
	ks:vset("a", { 1, 2, 3 })
	ks:vset("b", { 4, 5, 6 })
	ks:set("plain", "x")

	-- vdim still resolves, but the \x00-prefixed record does not surface
	testimony.assert_equal(3, ks:vdim())
	testimony.assert_equal(3, ks:count())
	local seen = {}
	for k in ks:scan() do
		seen[k] = true
	end
	testimony.assert_true(seen["a"] and seen["b"] and seen["plain"])
	testimony.assert_nil(seen["\x00vdim"])

	db:close()
	std.fs.remove(path)
end)

testify:that("tampering with an encrypted value's type tag fails authentication", function()
	local seed = litls.random_bytes(32)
	local pubkey = litls.ed25519_keypair(seed)
	local enc = { seed = seed, pubkey = pubkey }
	local path = tmp_path()

	local db = mneme.open(path, { sync = "none", encryption = enc })
	local sec = db:keyspace("s", { encrypted = true })
	sec:set_integer("n", 42)

	-- Reach under the API to rewrite the stored ciphertext with a forged
	-- type tag (INTEGER -> BYTES), keeping the value bytes intact. The tag
	-- is part of the AEAD's AAD, so the read must fail authentication
	-- rather than reinterpret the plaintext.
	local handle = db.__state.handle
	local txn = handle:txn_begin(false)
	local dir_root = txn:meta().keyspace_root_pgno
	local ks_root = core.unpack_u32((txn:btree_get(dir_root, "s")))
	local raw = select(1, txn:btree_get(ks_root, "n"))
	local new_ks_root = txn:btree_put(ks_root, "n", raw, 1, 0) -- tag 1 = BYTES
	local new_dir_root = txn:btree_put(dir_root, "s", core.pack_u32(new_ks_root), 1, 0)
	txn:set_ks_root(new_dir_root)
	txn:commit()

	local v, err = sec:get("n")
	testimony.assert_nil(v)
	testimony.assert_true(tostring(err):find("authentication") ~= nil, tostring(err))

	db:close()
	std.fs.remove(path)
end)

testify:that("persist on an encrypted keyspace re-encrypts under the new TTL", function()
	local seed = litls.random_bytes(32)
	local pubkey = litls.ed25519_keypair(seed)
	local enc = { seed = seed, pubkey = pubkey }
	local path = tmp_path()

	local db = mneme.open(path, { sync = "none", encryption = enc })
	local sec = db:keyspace("s", { encrypted = true })
	sec:set("tok", "secret", { ttl = 3600 })
	testimony.assert_not_nil(sec:ttl("tok"))

	testimony.assert_true(sec:persist("tok") == true)
	testimony.assert_nil(sec:ttl("tok"))
	-- The value must still decrypt after the TTL changed (AAD includes TTL)
	testimony.assert_equal("secret", sec:get("tok"))

	db:close()
	std.fs.remove(path)
end)

testify:that("structure names must not contain NUL bytes", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("t")

	local zs, zerr = ks:sorted_set("evil\x00\x02")
	testimony.assert_nil(zs)
	testimony.assert_equal("sorted set name must not contain null bytes", zerr)

	local lst, lerr = ks:list("evil\x00\x00")
	testimony.assert_nil(lst)
	testimony.assert_equal("list name must not contain null bytes", lerr)

	db:close()
	std.fs.remove(path)
end)

testify:that("set_integer and incr validate their numbers", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("t")

	testimony.assert_nil(ks:set_integer("n", 0 / 0))
	testimony.assert_nil(ks:set_integer("n", 1.5))
	testimony.assert_nil(ks:incr("c", 0.5))
	testimony.assert_nil((ks:set("k", "v", { ttl = "soon" })))

	-- An expired non-integer key behaves as missing for incr
	ks:set("expired", "bytes", { ttl = -10 })
	local v, err = ks:incr("expired")
	testimony.assert_equal(1, v, err)

	db:close()
	std.fs.remove(path)
end)

-- ── Sorted set semantics ──────────────────────────────────────────────

testify:that("range_by_rank supports negative ranks", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local zs = db:keyspace("t"):sorted_set("board")
	zs:add(10, "a", 20, "b", 30, "c")

	local all = zs:range_by_rank(0, -1)
	testimony.assert_equal(3, #all)
	testimony.assert_equal("a", all[1].member)
	testimony.assert_equal("c", all[3].member)

	local last_two = zs:range_by_rank(-2, -1)
	testimony.assert_equal(2, #last_two)
	testimony.assert_equal("b", last_two[1].member)

	testimony.assert_equal(0, #zs:range_by_rank(5, 10))

	db:close()
	std.fs.remove(path)
end)

testify:that("zs:iter honors min/max score bounds", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local zs = db:keyspace("t"):sorted_set("board")
	zs:add(10, "a", 20, "b", 30, "c", 40, "d")

	local got = {}
	for score, member in zs:iter({ min = 15, max = 35 }) do
		got[#got + 1] = member .. "=" .. score
	end
	testimony.assert_equal("b=20,c=30", table.concat(got, ","))

	got = {}
	for _, member in zs:iter({ min = 20, max = 40, reverse = true }) do
		got[#got + 1] = member
	end
	testimony.assert_equal("d,c,b", table.concat(got, ","))

	db:close()
	std.fs.remove(path)
end)

testify:that("zs:add update failure cannot duplicate score entries", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local zs = db:keyspace("t"):sorted_set("board")

	zs:add(10, "a")
	zs:add(99, "a") -- update: old score-index entry must be removed
	local all = zs:range_by_rank(0, -1)
	testimony.assert_equal(1, #all)
	testimony.assert_equal(99, all[1].score)
	testimony.assert_equal(1, zs:card())

	db:close()
	std.fs.remove(path)
end)

-- ── FT pause persistence ──────────────────────────────────────────────

testify:that("FT paused state is visible to other handles and survives reopen", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ft = db:ft_keyspace("c")
	ft:put("d1", "hello world")
	ft:pause_indexing()
	ft:put("d2", "bulk doc")

	-- A second handle on the same keyspace must refuse to search
	local ft2 = db:ft_keyspace("c")
	local r, err = ft2:search("hello")
	testimony.assert_nil(r)
	testimony.assert_equal("indexing paused", err)
	-- ...and deleting an unindexed paused doc must not be "corruption"
	testimony.assert_true(ft2:del("d2") == true)

	db:close()

	-- Reopen: the marker is persisted, search still refuses
	local db2 = mneme.open(path, { sync = "none" })
	local ft3 = db2:ft_keyspace("c")
	local r2, err2 = ft3:search("hello")
	testimony.assert_nil(r2)
	testimony.assert_equal("indexing paused", err2)

	testimony.assert_true(ft3:rebuild_index() == true)
	local results = ft3:search("hello")
	testimony.assert_equal(1, #results)

	db2:close()
	std.fs.remove(path)
end)

testify:that("ft_search survives oversized custom-analyzer query terms", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ft = db:ft_keyspace("c", {
		analyzer = function(text)
			if #text > 1000 then
				-- A term long enough to wrap a uint16 length field
				return { string.rep("x", 65525) }
			end
			return { "hello", "world" }
		end,
		analyzer_name = "oversize",
	})
	ft:put("d1", "short text")

	local results = ft:search(string.rep("q", 2000), { top_k = 5 })
	testimony.assert_equal(0, #results)

	db:close()
	std.fs.remove(path)
end)

-- ── Per-keyspace encryption marker ────────────────────────────────────

testify:that("keyspace encryption mode is persisted and enforced", function()
	local path = tmp_path()
	local seed = litls.random_bytes(32)
	local pubkey = litls.ed25519_keypair(seed)
	local enc = { seed = seed, pubkey = pubkey }

	local db = mneme.open(path, { sync = "none", encryption = enc })
	local sec = db:keyspace("secrets", { encrypted = true })
	sec:set("api", "sk-123")

	local plain, perr = db:keyspace("secrets")
	testimony.assert_nil(plain)
	testimony.assert_equal("keyspace is encrypted; pass encrypted = true", perr)

	db:keyspace("cache"):set("k", "v")
	local wrong = db:keyspace("cache", { encrypted = true })
	testimony.assert_nil(wrong)

	db:close()

	-- Markers survive reopen
	local db2 = mneme.open(path, { encryption = enc })
	testimony.assert_nil(db2:keyspace("secrets"))
	local sec2 = db2:keyspace("secrets", { encrypted = true })
	testimony.assert_equal("sk-123", sec2:get("api"))

	-- drop_keyspace clears the marker so the name can be reused plaintext
	testimony.assert_true(db2:drop_keyspace("secrets") == true)
	testimony.assert_not_nil(db2:keyspace("secrets"))

	db2:close()
	std.fs.remove(path)
end)

testify:that("encryption init on a non-empty database requires init flag", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	db:keyspace("data"):set("k", "v")
	db:close()

	local seed = litls.random_bytes(32)
	local pubkey = litls.ed25519_keypair(seed)

	local refused, err = mneme.open(path, { encryption = { seed = seed, pubkey = pubkey } })
	testimony.assert_nil(refused)
	testimony.assert_true(tostring(err):find("init = true") ~= nil)

	local db2 = mneme.open(path, { encryption = { seed = seed, pubkey = pubkey, init = true } })
	testimony.assert_not_nil(db2)
	db2:close()
	std.fs.remove(path)
end)

-- ── Vector guards ─────────────────────────────────────────────────────

testify:that("vsearch reports query dimension mismatches", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("v")
	ks:vset("a", { 1, 2, 3 })

	local r, err = ks:vsearch({ 1, 2 }, { top_k = 5 })
	testimony.assert_nil(r)
	testimony.assert_equal("dimension mismatch", err)

	-- vset over a non-vector key is refused
	ks:set("plain", "text")
	local ok2, err2 = ks:vset("plain", { 1, 2, 3 })
	testimony.assert_nil(ok2)
	testimony.assert_equal("type mismatch: key is not a vector", err2)

	db:close()
	std.fs.remove(path)
end)

-- ── Page accounting under churn ───────────────────────────────────────

testify:that("page accounting stays consistent under heavy retire churn", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local ks = db:keyspace("churn")

	-- Enough single-txn deletes to exceed one retire batch (480 pgnos)
	-- and force chunked persists with mid-loop appends.
	local big = string.rep("x", 3000) -- overflow value: extra page per key
	for round = 1, 3 do
		local ok = ks:batch(function(b)
			for i = 1, 600 do
				b:set("k" .. i, big .. round)
			end
		end)
		testimony.assert_true(ok == true)
	end

	local stats = db:stats()
	testimony.assert_equal(stats.page_count, stats.live_pages + stats.reclaimable_pages)
	testimony.assert_true(stats.retired_pages <= stats.page_count)

	db:close()
	std.fs.remove(path)
end)

testify:conclude()