gamedev-the-steward/simulation/regional/RegionalChunkedFileStore.gd

class_name RegionalChunkedFileStore
extends RefCounted

const SCHEMA_VERSION := 1
const FORMAT_ID := "regional_chunked_file_store"
const GENERATIONS_DIRECTORY := "generations"
const MANIFEST_FILE_NAME := "manifest.bin"
const CURRENT_POINTER_FILE_NAME := "current.bin"
const PREVIOUS_POINTER_FILE_NAME := "current.previous.bin"
const TEMP_POINTER_FILE_NAME := "current.next.bin"
const GENERATION_PREFIX := "generation_"
const GENERATION_DIGITS := 20
const MAX_BASE_PATH_LENGTH := 2048
const MAX_CHUNK_ID_LENGTH := 256
const MAX_CHUNKS := 100_000
const MAX_GENERATIONS := 4096
const MAX_MANIFEST_BYTES := 8 * 1024 * 1024
const MAX_CHUNK_BYTES := 32 * 1024 * 1024
const MAX_TOTAL_GENERATION_BYTES := 64 * 1024 * 1024
const MAX_POINTER_BYTES := 4096

var _base_directory: String
var _absolute_base_directory := ""
var _last_error := ""
var _last_telemetry: Dictionary = {}


func _init(base_directory: String) -> void:
	_base_directory = base_directory
	_absolute_base_directory = ProjectSettings.globalize_path(base_directory).simplify_path()
	_begin_telemetry("idle")


func save(archive: RegionalChunkedPersistence) -> bool:
	_last_error = ""
	_begin_telemetry("save")
	if not _base_directory_is_safe():
		return _fail("The caller-supplied regional save directory is unsafe")
	if archive == null:
		return _fail("Regional chunk archive is missing")
	var validation_started := Time.get_ticks_usec()
	var archive_bundle := archive.to_dictionary()
	var revalidated := RegionalChunkedPersistence.from_dictionary(archive_bundle)
	_add_validation_time(validation_started)
	if revalidated == null or revalidated.checksum() != archive.checksum():
		return _fail("Regional chunk archive failed pre-write validation")
	var chunk_ids := archive.get_chunk_ids()
	if chunk_ids.is_empty() or chunk_ids.size() > MAX_CHUNKS:
		return _fail("Regional chunk count exceeds the filesystem contract")
	if not _ensure_directories():
		return false
	var generation_ids := _list_generation_ids()
	if generation_ids.size() >= MAX_GENERATIONS:
		return _fail("Regional generation limit reached; explicit maintenance is required")
	var generation_id := _next_generation_id(generation_ids)
	if generation_id.is_empty():
		return _fail("Could not allocate a bounded regional generation ID")
	var final_generation_path := _generation_path(generation_id)
	var temporary_generation_path := _allocate_temporary_generation_path(generation_id)
	if temporary_generation_path.is_empty():
		return _fail("Could not allocate a temporary regional generation directory")
	var directory_error := DirAccess.make_dir_recursive_absolute(temporary_generation_path)
	if directory_error != OK and directory_error != ERR_ALREADY_EXISTS:
		return _fail("Could not create the temporary regional generation directory")

	var serialized_chunks: Array[PackedByteArray] = []
	var chunk_files: Array[Dictionary] = []
	var generation_bytes := 0
	var archive_manifest := archive.get_manifest()
	var archive_descriptors: Array = archive_manifest["chunks"]
	for index in range(chunk_ids.size()):
		var chunk_id := chunk_ids[index]
		if chunk_id.length() > MAX_CHUNK_ID_LENGTH:
			return _fail("Regional chunk ID exceeds the bounded file-store contract")
		var chunk := archive.get_chunk(StringName(chunk_id))
		if chunk.is_empty() or not _is_primitive_tree(chunk):
			return _fail("Regional chunk payload is not primitive-only")
		var serialized := var_to_bytes(chunk)
		if serialized.is_empty() or serialized.size() > MAX_CHUNK_BYTES:
			return _fail("Regional chunk exceeds the per-file size limit")
		generation_bytes += serialized.size()
		if generation_bytes > MAX_TOTAL_GENERATION_BYTES:
			return _fail("Regional generation exceeds the total size limit")
		serialized_chunks.append(serialized)
		var archive_descriptor: Dictionary = archive_descriptors[index]
		(
			chunk_files
			. append(
				{
					"chunk_id": chunk_id,
					"file_name": _chunk_file_name(index),
					"serialized_bytes": serialized.size(),
					"file_checksum": _bytes_checksum(serialized),
					"canonical_checksum": String(archive_descriptor["checksum"]),
				}
			)
		)
		_last_telemetry["serialized_bytes"] += serialized.size()

	var generation_manifest := {
		"schema_version": SCHEMA_VERSION,
		"format_id": FORMAT_ID,
		"generation_id": generation_id,
		"archive_checksum": archive.checksum(),
		"envelope_checksum": String(archive_manifest["envelope_checksum"]),
		"archive_manifest": archive_manifest,
		"chunk_files": chunk_files,
	}
	if not _is_primitive_tree(generation_manifest):
		return _fail("Regional generation manifest is not primitive-only")
	var manifest_bytes := var_to_bytes(generation_manifest)
	if manifest_bytes.is_empty() or manifest_bytes.size() > MAX_MANIFEST_BYTES:
		return _fail("Regional generation manifest exceeds the size limit")
	generation_bytes += manifest_bytes.size()
	if generation_bytes > MAX_TOTAL_GENERATION_BYTES:
		return _fail("Regional generation exceeds the total size limit")
	_last_telemetry["serialized_bytes"] += manifest_bytes.size()

	for index in range(serialized_chunks.size()):
		var chunk_path := temporary_generation_path.path_join(_chunk_file_name(index))
		if not _write_bytes(chunk_path, serialized_chunks[index]):
			return _fail("Could not write regional chunk file %d" % index)
	var manifest_path := temporary_generation_path.path_join(MANIFEST_FILE_NAME)
	if not _write_bytes(manifest_path, manifest_bytes):
		return _fail("Could not write the regional generation manifest")
	var readback := _read_generation_at_path(temporary_generation_path, generation_id, "")
	if readback == null or readback.checksum() != archive.checksum():
		return _fail("Temporary regional generation failed exact readback validation")
	if DirAccess.rename_absolute(temporary_generation_path, final_generation_path) != OK:
		return _fail("Could not atomically install the regional generation")
	var pointer := {
		"schema_version": SCHEMA_VERSION,
		"format_id": FORMAT_ID,
		"generation_id": generation_id,
		"manifest_checksum": _bytes_checksum(manifest_bytes),
		"archive_checksum": archive.checksum(),
	}
	if not _install_current_pointer(pointer):
		return false
	_last_telemetry["generation_id"] = generation_id
	_last_telemetry["succeeded"] = true
	return true


func load_archive() -> RegionalChunkedPersistence:
	_last_error = ""
	_begin_telemetry("load_archive")
	if not _base_directory_is_safe():
		_fail("The caller-supplied regional save directory is unsafe")
		return null
	var pointer := _read_current_pointer()
	var pointer_generation_id := String(pointer.get("generation_id", ""))
	var pointer_manifest_checksum := String(pointer.get("manifest_checksum", ""))
	var pointer_archive_checksum := String(pointer.get("archive_checksum", ""))
	var generation_ids := _list_generation_ids()
	generation_ids.reverse()
	for generation_id: String in generation_ids:
		var expected_manifest_checksum := (
			pointer_manifest_checksum if generation_id == pointer_generation_id else ""
		)
		var archive := _read_generation(generation_id, expected_manifest_checksum)
		if archive == null:
			continue
		_last_telemetry["generation_id"] = generation_id
		_last_telemetry["recovered_generation"] = (
			generation_id != pointer_generation_id
			or pointer_archive_checksum != archive.checksum()
			or _last_telemetry["pointer_mismatch"]
		)
		_last_telemetry["succeeded"] = true
		return archive
	_fail("No valid regional generation could be recovered")
	return null


func load_active_location_metadata(location_id: StringName) -> Dictionary:
	_last_error = ""
	_begin_telemetry("load_active_location_metadata")
	_last_telemetry["partial_load"] = true
	if not _base_directory_is_safe() or location_id.is_empty():
		_fail("Regional active-location request is invalid")
		return {}
	var pointer := _read_current_pointer()
	var pointer_generation_id := String(pointer.get("generation_id", ""))
	var pointer_manifest_checksum := String(pointer.get("manifest_checksum", ""))
	var pointer_archive_checksum := String(pointer.get("archive_checksum", ""))
	var generation_ids := _list_generation_ids()
	generation_ids.reverse()
	for generation_id: String in generation_ids:
		var expected_manifest_checksum := (
			pointer_manifest_checksum if generation_id == pointer_generation_id else ""
		)
		var partial := _read_active_location_at_generation(
			generation_id, location_id, expected_manifest_checksum
		)
		if bool(partial.get("location_absent", false)):
			_fail("The latest recoverable regional generation does not contain the location")
			return {}
		if partial.is_empty():
			continue
		_last_telemetry["generation_id"] = generation_id
		_last_telemetry["recovered_generation"] = (
			generation_id != pointer_generation_id
			or pointer_archive_checksum != String(partial["archive_checksum"])
			or _last_telemetry["pointer_mismatch"]
		)
		_last_telemetry["succeeded"] = true
		return partial
	_fail("No regional generation contains valid metadata for the active location")
	return {}


func prune_old_generations(keep_valid_generations: int = 2) -> int:
	_last_error = ""
	_begin_telemetry("prune_old_generations")
	if not _base_directory_is_safe():
		_fail("The caller-supplied regional save directory is unsafe")
		return 0
	if keep_valid_generations < 2:
		_fail("Conservative pruning must keep at least two valid generations")
		return 0
	var pointer := _read_current_pointer()
	var pointer_generation_id := String(pointer.get("generation_id", ""))
	var valid_generations: Array[Dictionary] = []
	var generation_ids := _list_generation_ids()
	generation_ids.reverse()
	for generation_id: String in generation_ids:
		var archive := _read_generation(generation_id, "")
		if archive != null:
			(
				valid_generations
				. append(
					{
						"generation_id": generation_id,
						"archive": archive,
					}
				)
			)
	var protected_ids: Dictionary = {}
	for index in range(mini(keep_valid_generations, valid_generations.size())):
		protected_ids[String(valid_generations[index]["generation_id"])] = true
	if not pointer_generation_id.is_empty():
		protected_ids[pointer_generation_id] = true
	var pruned := 0
	for candidate: Dictionary in valid_generations:
		var generation_id := String(candidate["generation_id"])
		if protected_ids.has(generation_id):
			continue
		var archive: RegionalChunkedPersistence = candidate["archive"]
		if not _remove_valid_generation(generation_id, archive):
			_fail("Conservative regional generation pruning stopped on an unexpected layout")
			_last_telemetry["generations_pruned"] = pruned
			return pruned
		pruned += 1
	_last_telemetry["generations_pruned"] = pruned
	_last_telemetry["succeeded"] = true
	return pruned


func get_base_directory() -> String:
	return _base_directory


func get_generations_directory() -> String:
	return _base_directory.path_join(GENERATIONS_DIRECTORY)


func get_current_pointer_path() -> String:
	return _base_directory.path_join(CURRENT_POINTER_FILE_NAME)


func get_generation_ids() -> Array[String]:
	return _list_generation_ids() if _base_directory_is_safe() else []


func get_generation_directory(generation_id: String) -> String:
	if not _is_generation_id(generation_id):
		return ""
	return get_generations_directory().path_join(generation_id)


func get_last_error() -> String:
	return _last_error


func get_last_telemetry() -> Dictionary:
	return _last_telemetry.duplicate(true)


func _read_generation(
	generation_id: String, expected_manifest_checksum: String
) -> RegionalChunkedPersistence:
	if not _is_generation_id(generation_id):
		return null
	return _read_generation_at_path(
		_generation_path(generation_id), generation_id, expected_manifest_checksum
	)


func _read_generation_at_path(
	generation_path: String, generation_id: String, expected_manifest_checksum: String
) -> RegionalChunkedPersistence:
	_last_telemetry["candidates_validated"] += 1
	var manifest_bytes := _read_bytes(
		generation_path.path_join(MANIFEST_FILE_NAME), MAX_MANIFEST_BYTES
	)
	if manifest_bytes.is_empty():
		return null
	if (
		not expected_manifest_checksum.is_empty()
		and _bytes_checksum(manifest_bytes) != expected_manifest_checksum
	):
		_last_telemetry["pointer_mismatch"] = true
	var validation_started := Time.get_ticks_usec()
	var decoded_manifest: Variant = bytes_to_var(manifest_bytes)
	if (
		not decoded_manifest is Dictionary
		or not _is_primitive_tree(decoded_manifest)
		or not _generation_manifest_is_valid(decoded_manifest, generation_id)
	):
		_add_validation_time(validation_started)
		return null
	var generation_manifest: Dictionary = decoded_manifest
	_add_validation_time(validation_started)
	var chunks: Array[Dictionary] = []
	var generation_bytes := manifest_bytes.size()
	for raw_entry: Variant in generation_manifest["chunk_files"]:
		var entry: Dictionary = raw_entry
		var chunk := _read_chunk_file(generation_path, entry)
		if chunk.is_empty():
			return null
		generation_bytes += int(entry["serialized_bytes"])
		if generation_bytes > MAX_TOTAL_GENERATION_BYTES:
			return null
		chunks.append(chunk)
	var archive_manifest: Dictionary = generation_manifest["archive_manifest"]
	var bundle := {
		"schema_version": RegionalChunkedPersistence.SCHEMA_VERSION,
		"manifest": archive_manifest.duplicate(true),
		"chunks": chunks,
	}
	validation_started = Time.get_ticks_usec()
	var archive := RegionalChunkedPersistence.from_dictionary(bundle)
	_add_validation_time(validation_started)
	if (
		archive == null
		or archive.checksum() != String(generation_manifest["archive_checksum"])
		or (
			String(archive.get_manifest()["envelope_checksum"])
			!= String(generation_manifest["envelope_checksum"])
		)
	):
		return null
	return archive


func _read_active_location_at_generation(
	generation_id: StringName, location_id: StringName, expected_manifest_checksum: String
) -> Dictionary:
	if not _is_generation_id(String(generation_id)):
		return {}
	_last_telemetry["candidates_validated"] += 1
	var generation_path := _generation_path(String(generation_id))
	var manifest_bytes := _read_bytes(
		generation_path.path_join(MANIFEST_FILE_NAME), MAX_MANIFEST_BYTES
	)
	if manifest_bytes.is_empty():
		return {}
	if (
		not expected_manifest_checksum.is_empty()
		and _bytes_checksum(manifest_bytes) != expected_manifest_checksum
	):
		_last_telemetry["pointer_mismatch"] = true
	var validation_started := Time.get_ticks_usec()
	var decoded_manifest: Variant = bytes_to_var(manifest_bytes)
	if (
		not decoded_manifest is Dictionary
		or not _is_primitive_tree(decoded_manifest)
		or not _generation_manifest_is_valid(decoded_manifest, String(generation_id))
	):
		_add_validation_time(validation_started)
		return {}
	var generation_manifest: Dictionary = decoded_manifest
	var archive_manifest: Dictionary = generation_manifest["archive_manifest"]
	var archive_descriptors: Array = archive_manifest["chunks"]
	var chunk_entries: Array = generation_manifest["chunk_files"]
	var location_chunk_id := "location:%s" % location_id
	var global_entry: Dictionary = {}
	var location_entry: Dictionary = {}
	for index in range(archive_descriptors.size()):
		var descriptor: Dictionary = archive_descriptors[index]
		if String(descriptor["chunk_id"]) == RegionalChunkedPersistence.GLOBAL_CHUNK_ID:
			global_entry = chunk_entries[index]
		elif String(descriptor["chunk_id"]) == location_chunk_id:
			location_entry = chunk_entries[index]
	_add_validation_time(validation_started)
	if global_entry.is_empty():
		return {}
	var global_chunk := _read_chunk_file(generation_path, global_entry)
	if (
		global_chunk.is_empty()
		or not _chunk_header_is_valid(global_chunk)
		or String(global_chunk.get("chunk_type", "")) != RegionalChunkedPersistence.CHUNK_GLOBAL
		or not global_chunk.get("payload") is Dictionary
	):
		return {}
	var global_payload: Dictionary = global_chunk["payload"]
	if not global_payload.get("location_ids") is Array:
		return {}
	if String(location_id) not in global_payload["location_ids"]:
		return {"location_absent": true}
	if location_entry.is_empty():
		return {}
	var location_chunk := _read_chunk_file(generation_path, location_entry)
	if (
		location_chunk.is_empty()
		or not _chunk_header_is_valid(location_chunk)
		or String(location_chunk.get("chunk_type", "")) != RegionalChunkedPersistence.CHUNK_LOCATION
		or String(location_chunk.get("scope_id", "")) != String(location_id)
		or not location_chunk.get("payload") is Dictionary
	):
		return {}
	var location_payload: Dictionary = location_chunk["payload"]
	if (
		String(global_payload.get("world_id", "")) != String(archive_manifest["world_id"])
		or String(location_payload.get("world_id", "")) != String(archive_manifest["world_id"])
		or String(location_payload.get("location_id", "")) != String(location_id)
	):
		return {}
	var loaded_chunk_ids: Array[String] = [
		RegionalChunkedPersistence.GLOBAL_CHUNK_ID, location_chunk_id
	]
	var unloaded_chunk_ids: Array[String] = []
	var mobile_group_refs: Array[Dictionary] = []
	for raw_descriptor: Variant in archive_descriptors:
		var descriptor: Dictionary = raw_descriptor
		var chunk_id := String(descriptor["chunk_id"])
		if chunk_id not in loaded_chunk_ids:
			unloaded_chunk_ids.append(chunk_id)
		if (
			String(descriptor["chunk_type"]) == RegionalChunkedPersistence.CHUNK_MOBILE_GROUP
			and String(descriptor["location_id"]) == String(location_id)
		):
			(
				mobile_group_refs
				. append(
					{
						"group_id": String(descriptor["scope_id"]),
						"chunk_id": chunk_id,
					}
				)
			)
	return {
		"schema_version": RegionalChunkedPersistence.SCHEMA_VERSION,
		"world_id": String(archive_manifest["world_id"]),
		"active_location_id": String(location_id),
		"loaded_chunk_ids": loaded_chunk_ids,
		"unloaded_chunk_ids": unloaded_chunk_ids,
		"global_index": global_payload.duplicate(true),
		"active_location": location_payload.duplicate(true),
		"mobile_group_refs": mobile_group_refs,
		"envelope_checksum": String(archive_manifest["envelope_checksum"]),
		"generation_id": String(generation_id),
		"archive_checksum": String(generation_manifest["archive_checksum"]),
		"full_generation_validated": false,
	}


func _read_chunk_file(generation_path: String, entry: Dictionary) -> Dictionary:
	var serialized_bytes := int(entry["serialized_bytes"])
	if serialized_bytes <= 0 or serialized_bytes > MAX_CHUNK_BYTES:
		return {}
	var chunk_bytes := _read_bytes(
		generation_path.path_join(String(entry["file_name"])), MAX_CHUNK_BYTES
	)
	if (
		chunk_bytes.size() != serialized_bytes
		or _bytes_checksum(chunk_bytes) != String(entry["file_checksum"])
	):
		return {}
	var validation_started := Time.get_ticks_usec()
	var decoded_chunk: Variant = bytes_to_var(chunk_bytes)
	if not decoded_chunk is Dictionary or not _is_primitive_tree(decoded_chunk):
		_add_validation_time(validation_started)
		return {}
	var chunk: Dictionary = decoded_chunk
	if (
		String(chunk.get("chunk_id", "")) != String(entry["chunk_id"])
		or JSON.stringify(chunk).sha256_text() != String(entry["canonical_checksum"])
	):
		_add_validation_time(validation_started)
		return {}
	_add_validation_time(validation_started)
	return chunk


func _generation_manifest_is_valid(manifest: Dictionary, generation_id: String) -> bool:
	var fields := [
		"schema_version",
		"format_id",
		"generation_id",
		"archive_checksum",
		"envelope_checksum",
		"archive_manifest",
		"chunk_files",
	]
	if manifest.size() != fields.size() or not manifest.has_all(fields):
		return false
	if (
		not manifest["schema_version"] is int
		or int(manifest["schema_version"]) != SCHEMA_VERSION
		or not manifest["format_id"] is String
		or String(manifest["format_id"]) != FORMAT_ID
		or not manifest["generation_id"] is String
		or String(manifest["generation_id"]) != generation_id
		or not _is_generation_id(generation_id)
		or not _is_sha256(manifest["archive_checksum"])
		or not _is_sha256(manifest["envelope_checksum"])
		or not manifest["archive_manifest"] is Dictionary
		or not manifest["chunk_files"] is Array
	):
		return false
	var archive_manifest: Dictionary = manifest["archive_manifest"]
	var archive_fields := [
		"schema_version",
		"format_id",
		"world_id",
		"envelope_checksum",
		"event_segment_size",
		"chunk_count",
		"chunks",
	]
	if (
		archive_manifest.size() != archive_fields.size()
		or not archive_manifest.has_all(archive_fields)
		or not archive_manifest["schema_version"] is int
		or not archive_manifest["format_id"] is String
		or not archive_manifest["world_id"] is String
		or String(archive_manifest["world_id"]).is_empty()
		or not archive_manifest["envelope_checksum"] is String
		or not archive_manifest["event_segment_size"] is int
		or not archive_manifest["chunks"] is Array
		or not archive_manifest["chunk_count"] is int
		or int(archive_manifest["chunk_count"]) <= 0
		or int(archive_manifest["chunk_count"]) > MAX_CHUNKS
		or (archive_manifest["chunks"] as Array).size() != int(archive_manifest["chunk_count"])
		or (manifest["chunk_files"] as Array).size() != int(archive_manifest["chunk_count"])
		or String(archive_manifest["format_id"]) != RegionalChunkedPersistence.FORMAT_ID
		or int(archive_manifest["schema_version"]) != RegionalChunkedPersistence.SCHEMA_VERSION
		or String(archive_manifest["envelope_checksum"]) != String(manifest["envelope_checksum"])
	):
		return false
	var ids: Dictionary = {}
	var archive_descriptors: Array = archive_manifest["chunks"]
	var chunk_files: Array = manifest["chunk_files"]
	for index in range(archive_descriptors.size()):
		var descriptor: Variant = archive_descriptors[index]
		var entry: Variant = chunk_files[index]
		if not descriptor is Dictionary or not entry is Dictionary:
			return false
		var descriptor_fields := [
			"chunk_id",
			"chunk_type",
			"scope_id",
			"location_id",
			"route_id",
			"record_count",
			"byte_size",
			"checksum",
		]
		var entry_fields := [
			"chunk_id",
			"file_name",
			"serialized_bytes",
			"file_checksum",
			"canonical_checksum",
		]
		if (
			descriptor.size() != descriptor_fields.size()
			or not descriptor.has_all(descriptor_fields)
			or entry.size() != entry_fields.size()
			or not entry.has_all(entry_fields)
		):
			return false
		var chunk_id := String(entry["chunk_id"])
		if (
			not descriptor["chunk_id"] is String
			or not descriptor["chunk_type"] is String
			or not descriptor["scope_id"] is String
			or not descriptor["location_id"] is String
			or not descriptor["route_id"] is String
			or not descriptor["record_count"] is int
			or not descriptor["byte_size"] is int
			or not descriptor["checksum"] is String
			or not entry["chunk_id"] is String
			or not entry["file_name"] is String
			or not entry["file_checksum"] is String
			or not entry["canonical_checksum"] is String
			or chunk_id.is_empty()
			or chunk_id.length() > MAX_CHUNK_ID_LENGTH
			or ids.has(chunk_id)
			or String(descriptor.get("chunk_id", "")) != chunk_id
			or String(entry["file_name"]) != _chunk_file_name(index)
			or not _is_safe_file_name(entry["file_name"])
			or not entry["serialized_bytes"] is int
			or int(entry["serialized_bytes"]) <= 0
			or int(entry["serialized_bytes"]) > MAX_CHUNK_BYTES
			or not _is_sha256(entry["file_checksum"])
			or not _is_sha256(entry["canonical_checksum"])
			or String(descriptor.get("checksum", "")) != String(entry["canonical_checksum"])
		):
			return false
		ids[chunk_id] = true
	return true


func _chunk_header_is_valid(chunk: Dictionary) -> bool:
	var fields := [
		"schema_version", "chunk_id", "chunk_type", "scope_id", "location_id", "route_id", "payload"
	]
	return (
		chunk.size() == fields.size()
		and chunk.has_all(fields)
		and chunk["schema_version"] is int
		and int(chunk["schema_version"]) == RegionalChunkedPersistence.SCHEMA_VERSION
		and chunk["chunk_id"] is String
		and chunk["chunk_type"] is String
		and chunk["scope_id"] is String
		and chunk["location_id"] is String
		and chunk["route_id"] is String
		and chunk["payload"] is Dictionary
	)


func _install_current_pointer(pointer: Dictionary) -> bool:
	if not _pointer_is_valid(pointer):
		return _fail("Regional current pointer is invalid")
	var pointer_bytes := var_to_bytes(pointer)
	if pointer_bytes.is_empty() or pointer_bytes.size() > MAX_POINTER_BYTES:
		return _fail("Regional current pointer exceeds the size limit")
	_last_telemetry["serialized_bytes"] += pointer_bytes.size()
	var temporary_path := _absolute_base_directory.path_join(TEMP_POINTER_FILE_NAME)
	var current_path := _absolute_base_directory.path_join(CURRENT_POINTER_FILE_NAME)
	var previous_path := _absolute_base_directory.path_join(PREVIOUS_POINTER_FILE_NAME)
	_remove_file_if_present(temporary_path)
	if not _write_bytes(temporary_path, pointer_bytes):
		return _fail("Could not write the temporary regional current pointer")
	var readback_bytes := _read_bytes(temporary_path, MAX_POINTER_BYTES)
	var readback: Variant = bytes_to_var(readback_bytes) if not readback_bytes.is_empty() else null
	if not readback is Dictionary or readback != pointer or not _pointer_is_valid(readback):
		return _fail("Regional current pointer failed readback validation")
	if DirAccess.rename_absolute(temporary_path, current_path) == OK:
		return true
	_remove_file_if_present(previous_path)
	var had_current := FileAccess.file_exists(current_path)
	if had_current and DirAccess.rename_absolute(current_path, previous_path) != OK:
		return _fail("Could not preserve the previous regional current pointer")
	if DirAccess.rename_absolute(temporary_path, current_path) == OK:
		return true
	if had_current:
		DirAccess.rename_absolute(previous_path, current_path)
	return _fail("Could not atomically install the regional current pointer")


func _read_current_pointer() -> Dictionary:
	var pointer_bytes := _read_bytes(
		_absolute_base_directory.path_join(CURRENT_POINTER_FILE_NAME), MAX_POINTER_BYTES
	)
	if pointer_bytes.is_empty():
		return {}
	var validation_started := Time.get_ticks_usec()
	var decoded_pointer: Variant = bytes_to_var(pointer_bytes)
	if (
		not decoded_pointer is Dictionary
		or not _is_primitive_tree(decoded_pointer)
		or not _pointer_is_valid(decoded_pointer)
	):
		_add_validation_time(validation_started)
		return {}
	_add_validation_time(validation_started)
	return decoded_pointer


func _pointer_is_valid(pointer: Dictionary) -> bool:
	var fields := [
		"schema_version", "format_id", "generation_id", "manifest_checksum", "archive_checksum"
	]
	return (
		pointer.size() == fields.size()
		and pointer.has_all(fields)
		and pointer["schema_version"] is int
		and int(pointer["schema_version"]) == SCHEMA_VERSION
		and pointer["format_id"] is String
		and String(pointer["format_id"]) == FORMAT_ID
		and pointer["generation_id"] is String
		and _is_generation_id(String(pointer["generation_id"]))
		and _is_sha256(pointer["manifest_checksum"])
		and _is_sha256(pointer["archive_checksum"])
	)


func _ensure_directories() -> bool:
	var base_error := DirAccess.make_dir_recursive_absolute(_absolute_base_directory)
	if base_error != OK and base_error != ERR_ALREADY_EXISTS:
		return _fail("Could not create the caller-supplied regional save directory")
	var generations_error := DirAccess.make_dir_recursive_absolute(_generations_path())
	if generations_error != OK and generations_error != ERR_ALREADY_EXISTS:
		return _fail("Could not create the regional generations directory")
	return true


func _base_directory_is_safe() -> bool:
	if (
		_base_directory.is_empty()
		or _base_directory.length() > MAX_BASE_PATH_LENGTH
		or _base_directory.to_utf8_buffer().has(0)
	):
		return false
	var normalized := _base_directory.replace("\\", "/")
	if normalized == "user://" or normalized == "res://":
		return false
	if not (
		normalized.begins_with("user://")
		or normalized.begins_with("res://")
		or normalized.is_absolute_path()
	):
		return false
	for segment: String in normalized.split("/", false):
		if segment == "." or segment == "..":
			return false
	if (
		_absolute_base_directory.is_empty()
		or not _absolute_base_directory.is_absolute_path()
		or _absolute_base_directory == "/"
		or _absolute_base_directory.get_file().is_empty()
	):
		return false
	return true


func _list_generation_ids() -> Array[String]:
	var ids: Array[String] = []
	var directory := DirAccess.open(_generations_path())
	if directory == null:
		return ids
	for raw_name: String in directory.get_directories():
		if _is_generation_id(raw_name):
			ids.append(raw_name)
	ids.sort()
	return ids


func _next_generation_id(existing_ids: Array[String]) -> String:
	var next_sequence := 1
	if not existing_ids.is_empty():
		var last_id := existing_ids[-1]
		var last_sequence_text := last_id.substr(GENERATION_PREFIX.length())
		var last_sequence := last_sequence_text.to_int()
		if last_sequence < 0 or last_sequence == 9_223_372_036_854_775_807:
			return ""
		next_sequence = last_sequence + 1
	return GENERATION_PREFIX + str(next_sequence).pad_zeros(GENERATION_DIGITS)


func _allocate_temporary_generation_path(generation_id: String) -> String:
	for attempt in range(16):
		var temporary_name := (
			"temporary_%s_%016d_%02d" % [generation_id, Time.get_ticks_usec(), attempt]
		)
		var candidate := _generations_path().path_join(temporary_name)
		if DirAccess.open(candidate) == null and not FileAccess.file_exists(candidate):
			return candidate
	return ""


func _is_generation_id(generation_id: String) -> bool:
	if generation_id.length() != GENERATION_PREFIX.length() + GENERATION_DIGITS:
		return false
	if not generation_id.begins_with(GENERATION_PREFIX):
		return false
	var sequence_text := generation_id.substr(GENERATION_PREFIX.length())
	for character: String in sequence_text:
		if character < "0" or character > "9":
			return false
	return sequence_text.to_int() > 0


func _chunk_file_name(index: int) -> String:
	return "chunk_%05d.bin" % index


func _is_safe_file_name(file_name: Variant) -> bool:
	if not file_name is String:
		return false
	var value := String(file_name)
	if value.is_empty() or value.length() > 64 or value.get_file() != value:
		return false
	var allowed := "abcdefghijklmnopqrstuvwxyz0123456789_.-"
	for character: String in value:
		if not allowed.contains(character):
			return false
	return true


func _is_sha256(value: Variant) -> bool:
	if not value is String or String(value).length() != 64:
		return false
	for character: String in String(value):
		if not "0123456789abcdef".contains(character):
			return false
	return true


func _write_bytes(path: String, bytes: PackedByteArray) -> bool:
	var started_usec := Time.get_ticks_usec()
	var file := FileAccess.open(path, FileAccess.WRITE)
	if file == null:
		_last_telemetry["write_usec"] += Time.get_ticks_usec() - started_usec
		return false
	file.store_buffer(bytes)
	file.flush()
	var write_error := file.get_error()
	file.close()
	_last_telemetry["write_usec"] += Time.get_ticks_usec() - started_usec
	if write_error != OK:
		return false
	_last_telemetry["bytes_written"] += bytes.size()
	_last_telemetry["files_written"] += 1
	return true


func _read_bytes(path: String, maximum_bytes: int) -> PackedByteArray:
	var started_usec := Time.get_ticks_usec()
	var file := FileAccess.open(path, FileAccess.READ)
	if file == null:
		_last_telemetry["read_usec"] += Time.get_ticks_usec() - started_usec
		return PackedByteArray()
	var length := file.get_length()
	if length <= 0 or length > maximum_bytes:
		file.close()
		_last_telemetry["read_usec"] += Time.get_ticks_usec() - started_usec
		return PackedByteArray()
	var bytes := file.get_buffer(length)
	var read_error := file.get_error()
	file.close()
	_last_telemetry["read_usec"] += Time.get_ticks_usec() - started_usec
	if read_error != OK or bytes.size() != length:
		return PackedByteArray()
	_last_telemetry["bytes_read"] += bytes.size()
	_last_telemetry["files_read"] += 1
	return bytes


func _bytes_checksum(bytes: PackedByteArray) -> String:
	var hashing_context := HashingContext.new()
	hashing_context.start(HashingContext.HASH_SHA256)
	hashing_context.update(bytes)
	return hashing_context.finish().hex_encode()


func _generations_path() -> String:
	return _absolute_base_directory.path_join(GENERATIONS_DIRECTORY)


func _generation_path(generation_id: String) -> String:
	return _generations_path().path_join(generation_id)


func _remove_file_if_present(path: String) -> void:
	if FileAccess.file_exists(path):
		DirAccess.remove_absolute(path)


func _remove_valid_generation(generation_id: String, archive: RegionalChunkedPersistence) -> bool:
	if not _is_generation_id(generation_id) or archive == null:
		return false
	var generation_path := _generation_path(generation_id)
	if not generation_path.begins_with(_generations_path() + "/"):
		return false
	var directory := DirAccess.open(generation_path)
	if directory == null:
		return false
	directory.include_hidden = true
	if not directory.get_directories().is_empty():
		return false
	var expected_files: Array[String] = [MANIFEST_FILE_NAME]
	for index in range(archive.get_chunk_ids().size()):
		expected_files.append(_chunk_file_name(index))
	expected_files.sort()
	var actual_files: Array[String] = []
	for file_name: String in directory.get_files():
		actual_files.append(file_name)
	actual_files.sort()
	if actual_files != expected_files:
		return false
	var started_usec := Time.get_ticks_usec()
	for file_name: String in actual_files:
		if DirAccess.remove_absolute(generation_path.path_join(file_name)) != OK:
			_last_telemetry["write_usec"] += Time.get_ticks_usec() - started_usec
			return false
		_last_telemetry["files_deleted"] += 1
	var remove_error := DirAccess.remove_absolute(generation_path)
	_last_telemetry["write_usec"] += Time.get_ticks_usec() - started_usec
	return remove_error == OK


func _begin_telemetry(operation: String) -> void:
	_last_telemetry = {
		"operation": operation,
		"generation_id": "",
		"serialized_bytes": 0,
		"bytes_written": 0,
		"bytes_read": 0,
		"files_written": 0,
		"files_read": 0,
		"files_deleted": 0,
		"write_usec": 0,
		"read_usec": 0,
		"validation_usec": 0,
		"candidates_validated": 0,
		"recovered_generation": false,
		"pointer_mismatch": false,
		"generations_pruned": 0,
		"partial_load": false,
		"succeeded": false,
	}


func _add_validation_time(started_usec: int) -> void:
	_last_telemetry["validation_usec"] += Time.get_ticks_usec() - started_usec


func _fail(message: String) -> bool:
	_last_error = message
	return false


static func _is_primitive_tree(value: Variant) -> bool:
	match typeof(value):
		TYPE_NIL, TYPE_BOOL, TYPE_INT, TYPE_STRING:
			return true
		TYPE_FLOAT:
			return is_finite(float(value))
		TYPE_ARRAY:
			for item: Variant in value:
				if not _is_primitive_tree(item):
					return false
			return true
		TYPE_DICTIONARY:
			for key: Variant in value:
				if not key is String or not _is_primitive_tree(value[key]):
					return false
			return true
	return false