Rijad Zuzo
487 lines
16 KiB
GDScript
487 lines
16 KiB
GDScript
class_name WorldEventRetentionPlan
|
|
extends RefCounted
|
|
|
|
const SCHEMA_VERSION := 1
|
|
const DECISION_RETAIN_RAW := "retain_raw"
|
|
const DECISION_ROLLED_UP := "rolled_up"
|
|
const RECORD_FIELDS := [
|
|
"schema_version",
|
|
"current_tick",
|
|
"ticks_per_day",
|
|
"raw_retention_days",
|
|
"first_raw_day_index",
|
|
"source_checksum",
|
|
"high_volume_event_types",
|
|
"pinned_fact_ids",
|
|
"causal_fact_ids",
|
|
"retained_event_ids",
|
|
"rolled_up_event_ids",
|
|
"rollup_store",
|
|
"reason_traces",
|
|
"telemetry",
|
|
]
|
|
const TRACE_FIELDS := [
|
|
"event_id",
|
|
"decision",
|
|
"reasons",
|
|
"cause_event_ids",
|
|
"required_by_event_ids",
|
|
"rollup_keys",
|
|
]
|
|
const TELEMETRY_FIELDS := [
|
|
"succeeded",
|
|
"input_event_count",
|
|
"retained_event_count",
|
|
"rolled_up_event_count",
|
|
"rollup_record_count",
|
|
"rollup_contribution_count",
|
|
"pinned_fact_count",
|
|
"causal_fact_count",
|
|
"causal_ancestor_count",
|
|
"recent_event_count",
|
|
"non_rollup_event_count",
|
|
"invalid_rollup_event_count",
|
|
"first_raw_day_index",
|
|
"source_checksum",
|
|
"source_unchanged",
|
|
]
|
|
|
|
var _data: Dictionary = {}
|
|
var _rollup_store := WorldEventDailyRollupStore.new()
|
|
var _traces_by_event_id: Dictionary = {}
|
|
|
|
|
|
static func create(
|
|
current_tick: int,
|
|
ticks_per_day: int,
|
|
raw_retention_days: int,
|
|
first_raw_day_index: int,
|
|
source_checksum: String,
|
|
high_volume_event_types: Array[StringName],
|
|
pinned_fact_ids: Array[int],
|
|
causal_fact_ids: Array[int],
|
|
retained_event_ids: Array[int],
|
|
rolled_up_event_ids: Array[int],
|
|
rollup_store: WorldEventDailyRollupStore,
|
|
reason_traces: Array[Dictionary],
|
|
telemetry: Dictionary
|
|
) -> WorldEventRetentionPlan:
|
|
if rollup_store == null:
|
|
return null
|
|
var serialized_event_types: Array[String] = []
|
|
for event_type: StringName in high_volume_event_types:
|
|
serialized_event_types.append(String(event_type))
|
|
return from_dictionary(
|
|
{
|
|
"schema_version": SCHEMA_VERSION,
|
|
"current_tick": current_tick,
|
|
"ticks_per_day": ticks_per_day,
|
|
"raw_retention_days": raw_retention_days,
|
|
"first_raw_day_index": first_raw_day_index,
|
|
"source_checksum": source_checksum,
|
|
"high_volume_event_types": serialized_event_types,
|
|
"pinned_fact_ids": pinned_fact_ids.duplicate(),
|
|
"causal_fact_ids": causal_fact_ids.duplicate(),
|
|
"retained_event_ids": retained_event_ids.duplicate(),
|
|
"rolled_up_event_ids": rolled_up_event_ids.duplicate(),
|
|
"rollup_store": rollup_store.to_dictionary(),
|
|
"reason_traces": reason_traces.duplicate(true),
|
|
"telemetry": telemetry.duplicate(true),
|
|
}
|
|
)
|
|
|
|
|
|
static func from_dictionary(record_data: Dictionary) -> WorldEventRetentionPlan:
|
|
if not _has_exact_fields(record_data, RECORD_FIELDS):
|
|
return null
|
|
if (
|
|
not record_data["schema_version"] is int
|
|
or int(record_data["schema_version"]) != SCHEMA_VERSION
|
|
or not record_data["current_tick"] is int
|
|
or not record_data["ticks_per_day"] is int
|
|
or not record_data["raw_retention_days"] is int
|
|
or not record_data["first_raw_day_index"] is int
|
|
or not record_data["source_checksum"] is String
|
|
or not record_data["high_volume_event_types"] is Array
|
|
or not record_data["pinned_fact_ids"] is Array
|
|
or not record_data["causal_fact_ids"] is Array
|
|
or not record_data["retained_event_ids"] is Array
|
|
or not record_data["rolled_up_event_ids"] is Array
|
|
or not record_data["rollup_store"] is Dictionary
|
|
or not record_data["reason_traces"] is Array
|
|
or not record_data["telemetry"] is Dictionary
|
|
):
|
|
return null
|
|
var event_types: Variant = _normalize_string_array(record_data["high_volume_event_types"])
|
|
var pinned_ids: Variant = _normalize_int_array(record_data["pinned_fact_ids"])
|
|
var causal_ids: Variant = _normalize_int_array(record_data["causal_fact_ids"])
|
|
var retained_ids: Variant = _normalize_int_array(record_data["retained_event_ids"])
|
|
var rolled_up_ids: Variant = _normalize_int_array(record_data["rolled_up_event_ids"])
|
|
if (
|
|
event_types == null
|
|
or pinned_ids == null
|
|
or causal_ids == null
|
|
or retained_ids == null
|
|
or rolled_up_ids == null
|
|
or event_types != record_data["high_volume_event_types"]
|
|
or pinned_ids != record_data["pinned_fact_ids"]
|
|
or causal_ids != record_data["causal_fact_ids"]
|
|
or retained_ids != record_data["retained_event_ids"]
|
|
or rolled_up_ids != record_data["rolled_up_event_ids"]
|
|
):
|
|
return null
|
|
var current_tick := int(record_data["current_tick"])
|
|
var ticks_per_day := int(record_data["ticks_per_day"])
|
|
var raw_retention_days := int(record_data["raw_retention_days"])
|
|
var first_raw_day_index := int(record_data["first_raw_day_index"])
|
|
if (
|
|
current_tick < 0
|
|
or ticks_per_day <= 0
|
|
or raw_retention_days < 0
|
|
or (first_raw_day_index != maxi(int(current_tick / ticks_per_day) - raw_retention_days, 0))
|
|
or not _is_sha256(String(record_data["source_checksum"]))
|
|
):
|
|
return null
|
|
var rollup_store := WorldEventDailyRollupStore.from_dictionary(record_data["rollup_store"])
|
|
if rollup_store == null:
|
|
return null
|
|
var normalized_traces: Variant = _normalize_traces(record_data["reason_traces"])
|
|
if normalized_traces == null or normalized_traces != record_data["reason_traces"]:
|
|
return null
|
|
var telemetry: Dictionary = record_data["telemetry"]
|
|
if not _telemetry_is_valid(telemetry, String(record_data["source_checksum"])):
|
|
return null
|
|
if not _cross_references_are_valid(
|
|
pinned_ids,
|
|
causal_ids,
|
|
retained_ids,
|
|
rolled_up_ids,
|
|
rollup_store,
|
|
normalized_traces,
|
|
telemetry,
|
|
first_raw_day_index
|
|
):
|
|
return null
|
|
var plan := WorldEventRetentionPlan.new()
|
|
plan._data = {
|
|
"current_tick": current_tick,
|
|
"ticks_per_day": ticks_per_day,
|
|
"raw_retention_days": raw_retention_days,
|
|
"first_raw_day_index": first_raw_day_index,
|
|
"source_checksum": String(record_data["source_checksum"]),
|
|
"high_volume_event_types": event_types,
|
|
"pinned_fact_ids": pinned_ids,
|
|
"causal_fact_ids": causal_ids,
|
|
"retained_event_ids": retained_ids,
|
|
"rolled_up_event_ids": rolled_up_ids,
|
|
"reason_traces": normalized_traces,
|
|
"telemetry": telemetry.duplicate(true),
|
|
}
|
|
plan._rollup_store = rollup_store
|
|
for trace: Dictionary in normalized_traces:
|
|
plan._traces_by_event_id[int(trace["event_id"])] = trace.duplicate(true)
|
|
return plan
|
|
|
|
|
|
func get_retained_event_ids() -> Array[int]:
|
|
return _copy_int_array(_data["retained_event_ids"])
|
|
|
|
|
|
func get_rolled_up_event_ids() -> Array[int]:
|
|
return _copy_int_array(_data["rolled_up_event_ids"])
|
|
|
|
|
|
func get_rollup_store() -> WorldEventDailyRollupStore:
|
|
return WorldEventDailyRollupStore.from_dictionary(_rollup_store.to_dictionary())
|
|
|
|
|
|
func get_reason_trace(event_id: int) -> Dictionary:
|
|
var trace: Dictionary = _traces_by_event_id.get(event_id, {} as Dictionary)
|
|
return trace.duplicate(true)
|
|
|
|
|
|
func get_reason_traces() -> Array[Dictionary]:
|
|
var traces: Array[Dictionary] = []
|
|
for trace: Dictionary in _data["reason_traces"]:
|
|
traces.append(trace.duplicate(true))
|
|
return traces
|
|
|
|
|
|
func get_telemetry() -> Dictionary:
|
|
return (_data["telemetry"] as Dictionary).duplicate(true)
|
|
|
|
|
|
func get_source_checksum() -> String:
|
|
return String(_data["source_checksum"])
|
|
|
|
|
|
func to_dictionary() -> Dictionary:
|
|
return {
|
|
"schema_version": SCHEMA_VERSION,
|
|
"current_tick": int(_data["current_tick"]),
|
|
"ticks_per_day": int(_data["ticks_per_day"]),
|
|
"raw_retention_days": int(_data["raw_retention_days"]),
|
|
"first_raw_day_index": int(_data["first_raw_day_index"]),
|
|
"source_checksum": String(_data["source_checksum"]),
|
|
"high_volume_event_types": (_data["high_volume_event_types"] as Array).duplicate(),
|
|
"pinned_fact_ids": (_data["pinned_fact_ids"] as Array).duplicate(),
|
|
"causal_fact_ids": (_data["causal_fact_ids"] as Array).duplicate(),
|
|
"retained_event_ids": (_data["retained_event_ids"] as Array).duplicate(),
|
|
"rolled_up_event_ids": (_data["rolled_up_event_ids"] as Array).duplicate(),
|
|
"rollup_store": _rollup_store.to_dictionary(),
|
|
"reason_traces": (_data["reason_traces"] as Array).duplicate(true),
|
|
"telemetry": (_data["telemetry"] as Dictionary).duplicate(true),
|
|
}
|
|
|
|
|
|
func checksum() -> String:
|
|
return JSON.stringify(to_dictionary()).sha256_text()
|
|
|
|
|
|
static func _normalize_traces(raw_traces: Array) -> Variant:
|
|
var traces: Array[Dictionary] = []
|
|
var seen: Dictionary = {}
|
|
var last_event_id := -1
|
|
for raw_trace: Variant in raw_traces:
|
|
if not raw_trace is Dictionary or not _has_exact_fields(raw_trace, TRACE_FIELDS):
|
|
return null
|
|
if (
|
|
not raw_trace["event_id"] is int
|
|
or not raw_trace["decision"] is String
|
|
or not raw_trace["reasons"] is Array
|
|
or not raw_trace["cause_event_ids"] is Array
|
|
or not raw_trace["required_by_event_ids"] is Array
|
|
or not raw_trace["rollup_keys"] is Array
|
|
):
|
|
return null
|
|
var event_id := int(raw_trace["event_id"])
|
|
var decision := String(raw_trace["decision"])
|
|
var reasons: Variant = _normalize_string_array(raw_trace["reasons"])
|
|
var cause_ids: Variant = _normalize_int_array(raw_trace["cause_event_ids"])
|
|
var required_by_ids: Variant = _normalize_int_array(raw_trace["required_by_event_ids"])
|
|
var rollup_keys: Variant = _normalize_string_array(raw_trace["rollup_keys"])
|
|
if (
|
|
event_id < 0
|
|
or event_id <= last_event_id
|
|
or seen.has(event_id)
|
|
or decision not in [DECISION_RETAIN_RAW, DECISION_ROLLED_UP]
|
|
or reasons == null
|
|
or reasons.is_empty()
|
|
or cause_ids == null
|
|
or required_by_ids == null
|
|
or rollup_keys == null
|
|
or reasons != raw_trace["reasons"]
|
|
or cause_ids != raw_trace["cause_event_ids"]
|
|
or required_by_ids != raw_trace["required_by_event_ids"]
|
|
or rollup_keys != raw_trace["rollup_keys"]
|
|
):
|
|
return null
|
|
seen[event_id] = true
|
|
last_event_id = event_id
|
|
traces.append(raw_trace.duplicate(true))
|
|
return traces
|
|
|
|
|
|
static func _telemetry_is_valid(telemetry: Dictionary, source_checksum: String) -> bool:
|
|
if not _has_exact_fields(telemetry, TELEMETRY_FIELDS):
|
|
return false
|
|
if (
|
|
telemetry["succeeded"] is not bool
|
|
or not bool(telemetry["succeeded"])
|
|
or telemetry["source_unchanged"] is not bool
|
|
or not bool(telemetry["source_unchanged"])
|
|
or telemetry["source_checksum"] is not String
|
|
or String(telemetry["source_checksum"]) != source_checksum
|
|
):
|
|
return false
|
|
for field: String in TELEMETRY_FIELDS:
|
|
if field in ["succeeded", "source_unchanged", "source_checksum"]:
|
|
continue
|
|
if telemetry[field] is not int or int(telemetry[field]) < 0:
|
|
return false
|
|
return true
|
|
|
|
|
|
static func _cross_references_are_valid(
|
|
pinned_ids: Array,
|
|
causal_ids: Array,
|
|
retained_ids: Array,
|
|
rolled_up_ids: Array,
|
|
rollup_store: WorldEventDailyRollupStore,
|
|
traces: Array,
|
|
telemetry: Dictionary,
|
|
first_raw_day_index: int
|
|
) -> bool:
|
|
var retained_set := _id_set(retained_ids)
|
|
var rolled_up_set := _id_set(rolled_up_ids)
|
|
var traces_by_id: Dictionary = {}
|
|
for trace: Dictionary in traces:
|
|
traces_by_id[int(trace["event_id"])] = trace
|
|
for event_id: int in retained_ids:
|
|
if rolled_up_set.has(event_id):
|
|
return false
|
|
for event_id: int in pinned_ids:
|
|
if (
|
|
not retained_set.has(event_id)
|
|
or not traces_by_id.has(event_id)
|
|
or "pinned_fact" not in (traces_by_id[event_id] as Dictionary)["reasons"]
|
|
):
|
|
return false
|
|
for event_id: int in causal_ids:
|
|
if (
|
|
not retained_set.has(event_id)
|
|
or not traces_by_id.has(event_id)
|
|
or "explicit_causal_fact" not in (traces_by_id[event_id] as Dictionary)["reasons"]
|
|
):
|
|
return false
|
|
var rollup_keys: Dictionary = {}
|
|
var contribution_count := 0
|
|
for record: WorldEventDailyRollupRecord in rollup_store.get_all():
|
|
rollup_keys[record.index_key()] = true
|
|
contribution_count += record.get_event_count()
|
|
var trace_ids: Dictionary = {}
|
|
var causal_ancestor_count := 0
|
|
var recent_count := 0
|
|
var non_rollup_count := 0
|
|
var invalid_rollup_count := 0
|
|
var source_ids_by_rollup_key: Dictionary = {}
|
|
for trace: Dictionary in traces:
|
|
var event_id := int(trace["event_id"])
|
|
trace_ids[event_id] = true
|
|
var is_retained := retained_set.has(event_id)
|
|
var is_rolled_up := rolled_up_set.has(event_id)
|
|
if is_retained == is_rolled_up:
|
|
return false
|
|
if is_retained and String(trace["decision"]) != DECISION_RETAIN_RAW:
|
|
return false
|
|
if is_rolled_up and String(trace["decision"]) != DECISION_ROLLED_UP:
|
|
return false
|
|
var reasons: Array = trace["reasons"]
|
|
var trace_rollup_keys: Array = trace["rollup_keys"]
|
|
if is_retained and not trace_rollup_keys.is_empty():
|
|
return false
|
|
if is_retained and "rolled_up_daily" in reasons:
|
|
return false
|
|
if is_rolled_up and (reasons != ["rolled_up_daily"] or trace_rollup_keys.is_empty()):
|
|
return false
|
|
for rollup_key: String in trace_rollup_keys:
|
|
if not rollup_keys.has(rollup_key):
|
|
return false
|
|
if not source_ids_by_rollup_key.has(rollup_key):
|
|
source_ids_by_rollup_key[rollup_key] = []
|
|
var source_ids: Array = source_ids_by_rollup_key[rollup_key]
|
|
source_ids.append(event_id)
|
|
for cause_id: int in trace["cause_event_ids"]:
|
|
if not retained_set.has(cause_id) or not traces_by_id.has(cause_id):
|
|
return false
|
|
var cause_trace: Dictionary = traces_by_id[cause_id]
|
|
if (
|
|
"causal_ancestor" not in cause_trace["reasons"]
|
|
or event_id not in cause_trace["required_by_event_ids"]
|
|
):
|
|
return false
|
|
for required_by_id: int in trace["required_by_event_ids"]:
|
|
if not traces_by_id.has(required_by_id):
|
|
return false
|
|
var required_by_trace: Dictionary = traces_by_id[required_by_id]
|
|
if event_id not in required_by_trace["cause_event_ids"]:
|
|
return false
|
|
if "causal_ancestor" in reasons:
|
|
causal_ancestor_count += 1
|
|
if "within_raw_retention" in reasons:
|
|
recent_count += 1
|
|
if "not_rollup_eligible" in reasons:
|
|
non_rollup_count += 1
|
|
if "rollup_payload_invalid" in reasons:
|
|
invalid_rollup_count += 1
|
|
if trace_ids.size() != retained_ids.size() + rolled_up_ids.size():
|
|
return false
|
|
if source_ids_by_rollup_key.size() != rollup_store.size():
|
|
return false
|
|
for record: WorldEventDailyRollupRecord in rollup_store.get_all():
|
|
var rollup_key := record.index_key()
|
|
if not source_ids_by_rollup_key.has(rollup_key):
|
|
return false
|
|
var source_ids: Array = source_ids_by_rollup_key[rollup_key]
|
|
source_ids.sort()
|
|
if (
|
|
record.get_event_count() != source_ids.size()
|
|
or record.get_minimum_event_id() != int(source_ids.front())
|
|
or record.get_maximum_event_id() != int(source_ids.back())
|
|
or record.get_source_event_ids_checksum() != JSON.stringify(source_ids).sha256_text()
|
|
):
|
|
return false
|
|
return (
|
|
int(telemetry["input_event_count"]) == trace_ids.size()
|
|
and int(telemetry["retained_event_count"]) == retained_ids.size()
|
|
and int(telemetry["rolled_up_event_count"]) == rolled_up_ids.size()
|
|
and int(telemetry["rollup_record_count"]) == rollup_store.size()
|
|
and int(telemetry["rollup_contribution_count"]) == contribution_count
|
|
and int(telemetry["pinned_fact_count"]) == pinned_ids.size()
|
|
and int(telemetry["causal_fact_count"]) == causal_ids.size()
|
|
and int(telemetry["causal_ancestor_count"]) == causal_ancestor_count
|
|
and int(telemetry["recent_event_count"]) == recent_count
|
|
and int(telemetry["non_rollup_event_count"]) == non_rollup_count
|
|
and int(telemetry["invalid_rollup_event_count"]) == invalid_rollup_count
|
|
and int(telemetry["first_raw_day_index"]) == first_raw_day_index
|
|
)
|
|
|
|
|
|
static func _normalize_string_array(values: Array) -> Variant:
|
|
var seen: Dictionary = {}
|
|
var normalized: Array[String] = []
|
|
for value: Variant in values:
|
|
if (not value is String and not value is StringName) or String(value).is_empty():
|
|
return null
|
|
var text := String(value)
|
|
if seen.has(text):
|
|
return null
|
|
seen[text] = true
|
|
normalized.append(text)
|
|
normalized.sort()
|
|
return normalized
|
|
|
|
|
|
static func _normalize_int_array(values: Array) -> Variant:
|
|
var seen: Dictionary = {}
|
|
var normalized: Array[int] = []
|
|
for value: Variant in values:
|
|
if not value is int or int(value) < 0 or seen.has(int(value)):
|
|
return null
|
|
seen[int(value)] = true
|
|
normalized.append(int(value))
|
|
normalized.sort()
|
|
return normalized
|
|
|
|
|
|
static func _id_set(values: Array) -> Dictionary:
|
|
var result: Dictionary = {}
|
|
for value: int in values:
|
|
result[value] = true
|
|
return result
|
|
|
|
|
|
static func _copy_int_array(values: Array) -> Array[int]:
|
|
var copied: Array[int] = []
|
|
for value: int in values:
|
|
copied.append(value)
|
|
return copied
|
|
|
|
|
|
static func _has_exact_fields(record_data: Dictionary, fields: Array) -> bool:
|
|
if record_data.size() != fields.size():
|
|
return false
|
|
for field: String in fields:
|
|
if not record_data.has(field):
|
|
return false
|
|
return true
|
|
|
|
|
|
static func _is_sha256(value: String) -> bool:
|
|
if value.length() != 64:
|
|
return false
|
|
for character_index in range(value.length()):
|
|
if value[character_index] not in "0123456789abcdef":
|
|
return false
|
|
return true
|