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

local testimony = require("testimony")
local std = require("std")
local mneme = require("mneme")

local testify = testimony.new("== mneme: B-tree operations ==")

local tmp_path = function()
	return "/tmp/mneme_btree_" .. std.nanoid() .. ".mneme"
end

-- Helper: open db, begin write txn, get initial root
local function setup()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })
	local txn = db:txn_begin(false)
	local info = db:info()
	return db, txn, info.keyspace_root_pgno, path
end

local function teardown(db, path)
	db:close()
	std.fs.remove(path)
end

testify:that("insert and get a single key-value", function()
	local db, txn, root, path = setup()

	root = txn:btree_put(root, "hello", "world")
	testimony.assert_not_nil(root)

	local val, tag, ttl = txn:btree_get(root, "hello")
	testimony.assert_equal("world", val)
	testimony.assert_equal(1, tag) -- MNEME_TYPE_BYTES

	txn:commit()
	teardown(db, path)
end)

testify:that("get non-existent key returns nil", function()
	local db, txn, root, path = setup()

	local val, err = txn:btree_get(root, "missing")
	testimony.assert_nil(val)
	testimony.assert_equal("not found", err)

	txn:abort()
	teardown(db, path)
end)

testify:that("insert 100 key-values and retrieve each", function()
	local db, txn, root, path = setup()

	for i = 1, 100 do
		local k = string.format("key:%04d", i)
		local v = string.format("value:%04d", i)
		root = txn:btree_put(root, k, v)
		testimony.assert_not_nil(root)
	end

	for i = 1, 100 do
		local k = string.format("key:%04d", i)
		local v = string.format("value:%04d", i)
		local got = txn:btree_get(root, k)
		testimony.assert_equal(v, got)
	end

	txn:commit()
	teardown(db, path)
end)

testify:that("insert keys in sorted (sequential) order", function()
	local db, txn, root, path = setup()

	for i = 1, 200 do
		local k = string.format("seq:%06d", i)
		root = txn:btree_put(root, k, "v" .. i)
		testimony.assert_not_nil(root)
	end

	-- Spot check
	testimony.assert_equal("v1", txn:btree_get(root, "seq:000001"))
	testimony.assert_equal("v100", txn:btree_get(root, "seq:000100"))
	testimony.assert_equal("v200", txn:btree_get(root, "seq:000200"))

	txn:commit()
	teardown(db, path)
end)

testify:that("insert keys in reverse order", function()
	local db, txn, root, path = setup()

	for i = 200, 1, -1 do
		local k = string.format("rev:%06d", i)
		root = txn:btree_put(root, k, "v" .. i)
		testimony.assert_not_nil(root)
	end

	testimony.assert_equal("v1", txn:btree_get(root, "rev:000001"))
	testimony.assert_equal("v200", txn:btree_get(root, "rev:000200"))

	txn:commit()
	teardown(db, path)
end)

testify:that("delete a key", function()
	local db, txn, root, path = setup()

	root = txn:btree_put(root, "a", "1")
	root = txn:btree_put(root, "b", "2")
	root = txn:btree_put(root, "c", "3")

	-- Delete b
	root = txn:btree_del(root, "b")
	testimony.assert_not_nil(root)

	-- b is gone
	local val, err = txn:btree_get(root, "b")
	testimony.assert_nil(val)

	-- a and c still exist
	testimony.assert_equal("1", txn:btree_get(root, "a"))
	testimony.assert_equal("3", txn:btree_get(root, "c"))

	txn:commit()
	teardown(db, path)
end)

testify:that("delete all keys one by one", function()
	local db, txn, root, path = setup()

	local keys = { "alpha", "beta", "gamma", "delta", "epsilon" }
	for _, k in ipairs(keys) do
		root = txn:btree_put(root, k, "val_" .. k)
	end

	for _, k in ipairs(keys) do
		root = txn:btree_del(root, k)
		local val = txn:btree_get(root, k)
		testimony.assert_nil(val)
	end

	txn:commit()
	teardown(db, path)
end)

testify:that("delete non-existent key returns same root", function()
	local db, txn, root, path = setup()

	root = txn:btree_put(root, "exists", "yes")
	local root_before = root
	root = txn:btree_del(root, "nope")
	testimony.assert_equal(root_before, root)

	txn:commit()
	teardown(db, path)
end)

testify:that("overwrite existing key with new value", function()
	local db, txn, root, path = setup()

	root = txn:btree_put(root, "key", "old_value")
	testimony.assert_equal("old_value", txn:btree_get(root, "key"))

	root = txn:btree_put(root, "key", "new_value")
	testimony.assert_equal("new_value", txn:btree_get(root, "key"))

	txn:commit()
	teardown(db, path)
end)

testify:that("insert triggers leaf splits (many keys)", function()
	local db, txn, root, path = setup()

	-- With ~45 byte cells, a leaf holds ~86. Insert 500 to force multiple splits.
	for i = 1, 500 do
		local k = string.format("split:%06d", i)
		local v = string.format("value:%06d", i)
		root = txn:btree_put(root, k, v)
		testimony.assert_not_nil(root)
	end

	-- Verify all keys
	for i = 1, 500 do
		local k = string.format("split:%06d", i)
		local v = string.format("value:%06d", i)
		local got = txn:btree_get(root, k)
		testimony.assert_equal(v, got)
	end

	txn:commit()
	teardown(db, path)
end)

testify:that("large value uses overflow pages", function()
	local db, txn, root, path = setup()

	-- Create a value larger than MAX_CELL_BODY (~2023 bytes)
	local large_val = string.rep("X", 5000)
	root = txn:btree_put(root, "big", large_val)
	testimony.assert_not_nil(root)

	local got = txn:btree_get(root, "big")
	testimony.assert_equal(5000, #got)
	testimony.assert_equal(large_val, got)

	txn:commit()
	teardown(db, path)
end)

testify:that("very large value (10KB) with overflow", function()
	local db, txn, root, path = setup()

	local huge_val = string.rep("A", 10000)
	root = txn:btree_put(root, "huge", huge_val)
	testimony.assert_not_nil(root)

	local got = txn:btree_get(root, "huge")
	testimony.assert_equal(10000, #got)
	testimony.assert_equal(huge_val, got)

	txn:commit()
	teardown(db, path)
end)

testify:that("overwrite overflow value frees old pages", function()
	local db, txn, root, path = setup()

	local v1 = string.rep("A", 5000)
	root = txn:btree_put(root, "ov", v1)

	local v2 = string.rep("B", 3000)
	root = txn:btree_put(root, "ov", v2)

	local got = txn:btree_get(root, "ov")
	testimony.assert_equal(3000, #got)
	testimony.assert_equal(v2, got)

	txn:commit()
	teardown(db, path)
end)

testify:that("data persists across transactions", function()
	local path = tmp_path()
	local db = mneme.open(path, { sync = "none" })

	-- Write in first txn
	local txn1 = db:txn_begin(false)
	local root = db:info().keyspace_root_pgno
	root = txn1:btree_put(root, "persist", "yes")
	txn1:commit()

	-- Read in second txn (using new root from db info after commit)
	-- Note: the root changed, but we need to track it.
	-- For Phase 2, we use the committed root by re-reading info.
	-- The keyspace_root_pgno in meta doesn't change automatically;
	-- that's Phase 3's job. For now, track root manually.
	local txn2 = db:txn_begin(false)
	local val = txn2:btree_get(root, "persist")
	testimony.assert_equal("yes", val)
	txn2:abort()

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

testify:that("type_tag is preserved", function()
	local db, txn, root, path = setup()

	root = txn:btree_put(root, "bytes", "hello", 1) -- Bytes
	root = txn:btree_put(root, "int", "42", 2) -- Integer
	root = txn:btree_put(root, "float", "3.14", 3) -- Float

	local _, tag1 = txn:btree_get(root, "bytes")
	local _, tag2 = txn:btree_get(root, "int")
	local _, tag3 = txn:btree_get(root, "float")

	testimony.assert_equal(1, tag1)
	testimony.assert_equal(2, tag2)
	testimony.assert_equal(3, tag3)

	txn:commit()
	teardown(db, path)
end)

testify:that("mixed insert and delete operations", function()
	local db, txn, root, path = setup()

	-- Insert 50 keys
	for i = 1, 50 do
		root = txn:btree_put(root, string.format("mix:%03d", i), "v" .. i)
	end

	-- Delete even-numbered keys
	for i = 2, 50, 2 do
		root = txn:btree_del(root, string.format("mix:%03d", i))
	end

	-- Verify: odd keys exist, even keys don't
	for i = 1, 50 do
		local k = string.format("mix:%03d", i)
		local val = txn:btree_get(root, k)
		if i % 2 == 1 then
			testimony.assert_equal("v" .. i, val)
		else
			testimony.assert_nil(val)
		end
	end

	txn:commit()
	teardown(db, path)
end)

testify:that("leaf split distributes VARIABLE-SIZE cells by bytes (no silent key loss)", function()
	-- Regression: the leaf split used to pick its split point by CELL COUNT.
	-- With large inline values a count split can pile more bytes on one side
	-- than a page holds; the overflowing cell_insert failed and the key was
	-- SILENTLY DROPPED while the put reported success. This exact shape (five
	-- keys, sizes mirroring a game autosave, "sim" inserted mid-order) lost
	-- the "world" key.
	local db, txn, root, path = setup()

	local sizes = { meta = 49, player = 222, world = 1720, ships = 1900, sim = 398 }
	for _, k in ipairs({ "meta", "player", "world", "ships", "sim" }) do
		root = txn:btree_put(root, k, string.rep("x", sizes[k]))
		testimony.assert_not_nil(root)
	end
	for k, n in pairs(sizes) do
		local val = txn:btree_get(root, k)
		testimony.assert_not_nil(val, "key survives the split: " .. k)
		testimony.assert_equal(n, #val)
	end

	txn:commit()
	teardown(db, path)
end)

testify:that("mid-order inserts of large values keep every key across splits", function()
	-- Sweep the same class of layouts: near-page-half values inserted in an
	-- order that forces byte-heavy mid splits at several sizes.
	for sz = 300, 2000, 170 do
		local db, txn, root, path = setup()
		local keys = { "a", "e", "b", "d", "c", "g", "f" }
		for i, k in ipairs(keys) do
			root = txn:btree_put(root, k, string.rep(k, (i % 2 == 0) and sz or 1900))
			testimony.assert_not_nil(root)
		end
		for i, k in ipairs(keys) do
			local val = txn:btree_get(root, k)
			testimony.assert_not_nil(val, string.format("sz=%d key=%s survives", sz, k))
			testimony.assert_equal((i % 2 == 0) and sz or 1900, #val)
		end
		txn:commit()
		teardown(db, path)
	end
end)

testify:conclude()