-- Engine adapter. dofile returns factory(schema, core, mod_storage).
-- Callers must fail startup on world errors and catch/kick on player load errors.
-- No logging of JSON, player identities, parser errors, or location data.
return function(schema, core, storage)
	local P = {}
	local world_kinds = {claims = true, homes = true, access = true, location = true, spawn = true}
	local blocked = {}
	local player_stores = setmetatable({}, {__mode = "k"})
	local function fail()
		error("Last Hearth: persistence failed; save retained for recovery", 0)
	end
	local function call(fn, ...)
		local ok, value, second = pcall(fn, ...)
		if not ok then fail() end
		return value, second
	end
	local function read(store, key)
		if blocked[store] and blocked[store][key] then fail() end
		local raw = call(store.get_string, store, key)
		if type(raw) ~= "string" then fail() end
		return raw
	end
	local function decode(kind, raw)
		local value = call(core.parse_json, raw, nil, true)
		if type(value) ~= "table" then fail() end
		return call(schema.decode, kind, value)
	end
	local function equal(a, b)
		if type(a) ~= type(b) then return false end
		if type(a) ~= "table" then return a == b end
		for k, v in pairs(a) do if not equal(v, b[k]) then return false end end
		for k in pairs(b) do if a[k] == nil then return false end end
		return true
	end
	local function encode(kind, envelope)
		local raw = call(core.write_json, envelope)
		if type(raw) ~= "string" or raw == "" then fail() end
		-- Validate the actual serialized result before touching either key.
		local roundtrip, migrated = decode(kind, raw)
		if migrated ~= false or not equal(envelope, roundtrip) then fail() end
		return raw
	end
	local function backup(store, key, original)
		-- Nonempty backup bytes are immutable, even if they are not valid JSON.
		if read(store, key) ~= "" then return end
		local result = call(store.set_string, store, key, original)
		if result == false or read(store, key) ~= original then fail() end
	end
	local function commit(store, key, original, raw)
		local ok = pcall(function()
			local result = call(store.set_string, store, key, raw)
			if result == false or read(store, key) ~= raw then fail() end
		end)
		if ok then return end
		-- Engine metadata offers no transaction/CAS or crash-durability guarantee.
		-- Best-effort byte-exact rollback must itself be verified. On failure the
		-- key is quarantined for this adapter's lifetime; caller must stop/kick.
		local restored = pcall(function()
			if read(store, key) ~= original then
				local result = call(store.set_string, store, key, original)
				if result == false then fail() end
			end
			if read(store, key) ~= original then fail() end
		end)
		if not restored then
			blocked[store] = blocked[store] or {}
			blocked[store][key] = true
		end
		fail()
	end
	local function load(store, key, backup_key, kind)
		local original = read(store, key)
		if original == "" then
			if kind == "needs" then return call(schema.defaults) end
			if kind == "claims" or kind == "homes" or kind == "access" then return {} end
			return nil
		end
		local envelope, migrated = decode(kind, original)
		if migrated then
			local raw = encode(kind, envelope)
			backup(store, backup_key, original)
			commit(store, key, original, raw)
		end
		return envelope.data
	end
	local function save(store, key, backup_key, kind, data)
		local envelope = call(schema.decode, kind, {schema = 1, kind = kind, data = data})
		local raw = encode(kind, envelope)
		local original = read(store, key)
		local migrated = false
		if original ~= "" then
			local previous, legacy = decode(kind, original)
			if not previous then fail() end
			migrated = legacy
		end
		if migrated then backup(store, backup_key, original) end
		commit(store, key, original, raw)
	end
	local function world_kind(kind)
		if type(kind) ~= "string" or not world_kinds[kind] then fail() end
	end
	local function metadata(player)
		-- get_meta() may return a fresh userdata wrapper. Keep the same reference
		-- for a player's session so failed-rollback quarantine cannot be bypassed.
		if player_stores[player] then return player_stores[player] end
		local meta = call(player.get_meta, player)
		if meta == nil then fail() end
		player_stores[player] = meta
		return meta
	end
	function P.load_world(kind)
		world_kind(kind)
		return load(storage, kind, "m0_backup:" .. kind, kind)
	end
	function P.save_world(kind, data)
		world_kind(kind)
		save(storage, kind, "m0_backup:" .. kind, kind, data)
	end
	function P.load_player(player)
		return load(metadata(player), "hearth:needs", "hearth:m0_backup:needs", "needs")
	end
	function P.save_player(player, needs)
		save(metadata(player), "hearth:needs", "hearth:m0_backup:needs", "needs", needs)
	end
	return P
end