feat: plan bounded event history rollups
This commit is contained in:
Rijad Zuzo
@@ -0,0 +1,215 @@
|
||||
class_name WorldEventDailyRollupStore
|
||||
extends RefCounted
|
||||
|
||||
const SCHEMA_VERSION := 1
|
||||
const RECORD_FIELDS := ["schema_version", "rollups"]
|
||||
|
||||
var _records_by_key: Dictionary = {}
|
||||
var _ordered_keys: Array[String] = []
|
||||
var _day_index: Dictionary = {}
|
||||
var _location_index: Dictionary = {}
|
||||
var _event_type_index: Dictionary = {}
|
||||
var _item_index: Dictionary = {}
|
||||
|
||||
|
||||
static func from_dictionary(store_data: Dictionary) -> WorldEventDailyRollupStore:
|
||||
var store := WorldEventDailyRollupStore.new()
|
||||
return store if store.restore_from_dictionary(store_data) else null
|
||||
|
||||
|
||||
func size() -> int:
|
||||
return _records_by_key.size()
|
||||
|
||||
|
||||
func is_empty() -> bool:
|
||||
return _records_by_key.is_empty()
|
||||
|
||||
|
||||
func has_rollup(
|
||||
day_index: int,
|
||||
world_id: StringName,
|
||||
location_id: StringName,
|
||||
event_type: StringName,
|
||||
item_id: StringName
|
||||
) -> bool:
|
||||
return _records_by_key.has(
|
||||
WorldEventDailyRollupRecord.index_key_for(
|
||||
day_index, world_id, location_id, event_type, item_id
|
||||
)
|
||||
)
|
||||
|
||||
|
||||
func append(record: WorldEventDailyRollupRecord) -> bool:
|
||||
return _append(record)
|
||||
|
||||
|
||||
func get_by_key(
|
||||
day_index: int,
|
||||
world_id: StringName,
|
||||
location_id: StringName,
|
||||
event_type: StringName,
|
||||
item_id: StringName
|
||||
) -> WorldEventDailyRollupRecord:
|
||||
var key := WorldEventDailyRollupRecord.index_key_for(
|
||||
day_index, world_id, location_id, event_type, item_id
|
||||
)
|
||||
return _records_by_key.get(key) as WorldEventDailyRollupRecord
|
||||
|
||||
|
||||
func get_all() -> Array[WorldEventDailyRollupRecord]:
|
||||
return _records_for_keys(_ordered_keys)
|
||||
|
||||
|
||||
func get_for_day(day_index: int) -> Array[WorldEventDailyRollupRecord]:
|
||||
return _records_for_keys(_day_index.get(day_index, []))
|
||||
|
||||
|
||||
func get_for_location(
|
||||
world_id: StringName, location_id: StringName
|
||||
) -> Array[WorldEventDailyRollupRecord]:
|
||||
return _records_for_keys(
|
||||
_location_index.get(WorldEventDailyRollupRecord.location_key_for(world_id, location_id), [])
|
||||
)
|
||||
|
||||
|
||||
func get_for_event_type(event_type: StringName) -> Array[WorldEventDailyRollupRecord]:
|
||||
return _records_for_keys(_event_type_index.get(event_type, []))
|
||||
|
||||
|
||||
func get_for_item(item_id: StringName) -> Array[WorldEventDailyRollupRecord]:
|
||||
return _records_for_keys(_item_index.get(item_id, []))
|
||||
|
||||
|
||||
func restore(records: Array[WorldEventDailyRollupRecord]) -> bool:
|
||||
var candidate := WorldEventDailyRollupStore.new()
|
||||
for record: WorldEventDailyRollupRecord in records:
|
||||
if not candidate._append(record):
|
||||
return false
|
||||
_adopt(candidate)
|
||||
return true
|
||||
|
||||
|
||||
func restore_from_dictionary(store_data: Dictionary) -> bool:
|
||||
if not _has_exact_fields(store_data, RECORD_FIELDS):
|
||||
return false
|
||||
if (
|
||||
not store_data["schema_version"] is int
|
||||
or int(store_data["schema_version"]) != SCHEMA_VERSION
|
||||
or not store_data["rollups"] is Array
|
||||
):
|
||||
return false
|
||||
var records: Array[WorldEventDailyRollupRecord] = []
|
||||
for raw_record: Variant in store_data["rollups"]:
|
||||
if not raw_record is Dictionary:
|
||||
return false
|
||||
var record := WorldEventDailyRollupRecord.from_dictionary(raw_record)
|
||||
if record == null:
|
||||
return false
|
||||
records.append(record)
|
||||
if not _records_are_canonical(records):
|
||||
return false
|
||||
return restore(records)
|
||||
|
||||
|
||||
func to_dictionary() -> Dictionary:
|
||||
var serialized: Array[Dictionary] = []
|
||||
for record: WorldEventDailyRollupRecord in get_all():
|
||||
serialized.append(record.to_dictionary())
|
||||
return {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"rollups": serialized,
|
||||
}
|
||||
|
||||
|
||||
func checksum() -> String:
|
||||
return JSON.stringify(to_dictionary()).sha256_text()
|
||||
|
||||
|
||||
func _append(record: WorldEventDailyRollupRecord) -> bool:
|
||||
if record == null or not record.is_valid() or _records_by_key.has(record.index_key()):
|
||||
return false
|
||||
var stored := WorldEventDailyRollupRecord.from_dictionary(record.to_dictionary())
|
||||
if stored == null:
|
||||
return false
|
||||
var key := stored.index_key()
|
||||
_records_by_key[key] = stored
|
||||
_insert_key(_ordered_keys, key)
|
||||
_index_key(_day_index, stored.get_day_index(), key)
|
||||
_index_key(_location_index, stored.location_index_key(), key)
|
||||
_index_key(_event_type_index, stored.get_event_type(), key)
|
||||
_index_key(_item_index, stored.get_item_id(), key)
|
||||
return true
|
||||
|
||||
|
||||
func _index_key(index: Dictionary, index_value: Variant, record_key: String) -> void:
|
||||
if not index.has(index_value):
|
||||
index[index_value] = []
|
||||
var keys: Array = index[index_value]
|
||||
_insert_key(keys, record_key)
|
||||
|
||||
|
||||
func _insert_key(keys: Array, record_key: String) -> void:
|
||||
var low := 0
|
||||
var high := keys.size()
|
||||
while low < high:
|
||||
var middle := (low + high) / 2
|
||||
if _record_precedes(record_key, String(keys[middle])):
|
||||
high = middle
|
||||
else:
|
||||
low = middle + 1
|
||||
keys.insert(low, record_key)
|
||||
|
||||
|
||||
func _record_precedes(first_key: String, second_key: String) -> bool:
|
||||
var first := _records_by_key[first_key] as WorldEventDailyRollupRecord
|
||||
var second := _records_by_key[second_key] as WorldEventDailyRollupRecord
|
||||
if first.get_day_index() != second.get_day_index():
|
||||
return first.get_day_index() < second.get_day_index()
|
||||
if first.get_world_id() != second.get_world_id():
|
||||
return String(first.get_world_id()) < String(second.get_world_id())
|
||||
if first.get_location_id() != second.get_location_id():
|
||||
return String(first.get_location_id()) < String(second.get_location_id())
|
||||
if first.get_event_type() != second.get_event_type():
|
||||
return String(first.get_event_type()) < String(second.get_event_type())
|
||||
return String(first.get_item_id()) < String(second.get_item_id())
|
||||
|
||||
|
||||
func _records_for_keys(keys: Array) -> Array[WorldEventDailyRollupRecord]:
|
||||
var records: Array[WorldEventDailyRollupRecord] = []
|
||||
for key: Variant in keys:
|
||||
var record := _records_by_key.get(String(key)) as WorldEventDailyRollupRecord
|
||||
if record != null:
|
||||
records.append(record)
|
||||
return records
|
||||
|
||||
|
||||
func _records_are_canonical(records: Array[WorldEventDailyRollupRecord]) -> bool:
|
||||
var candidate := WorldEventDailyRollupStore.new()
|
||||
for record: WorldEventDailyRollupRecord in records:
|
||||
if not candidate._append(record):
|
||||
return false
|
||||
var canonical_records := candidate.get_all()
|
||||
if canonical_records.size() != records.size():
|
||||
return false
|
||||
for index in range(records.size()):
|
||||
if canonical_records[index].index_key() != records[index].index_key():
|
||||
return false
|
||||
return true
|
||||
|
||||
|
||||
func _adopt(source: WorldEventDailyRollupStore) -> void:
|
||||
_records_by_key = source._records_by_key
|
||||
_ordered_keys = source._ordered_keys
|
||||
_day_index = source._day_index
|
||||
_location_index = source._location_index
|
||||
_event_type_index = source._event_type_index
|
||||
_item_index = source._item_index
|
||||
|
||||
|
||||
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
|
||||
@@ -0,0 +1 @@
|
||||
uid://p61id8fes4wk
|
||||
@@ -0,0 +1,486 @@
|
||||
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
|
||||
@@ -0,0 +1 @@
|
||||
uid://bj0uo56quas27
|
||||
@@ -0,0 +1,435 @@
|
||||
class_name WorldEventRetentionPlanner
|
||||
extends RefCounted
|
||||
|
||||
const REASON_PINNED_FACT := "pinned_fact"
|
||||
const REASON_EXPLICIT_CAUSAL_FACT := "explicit_causal_fact"
|
||||
const REASON_CAUSAL_ANCESTOR := "causal_ancestor"
|
||||
const REASON_WITHIN_RAW_RETENTION := "within_raw_retention"
|
||||
const REASON_NOT_ROLLUP_ELIGIBLE := "not_rollup_eligible"
|
||||
const REASON_ROLLUP_PAYLOAD_INVALID := "rollup_payload_invalid"
|
||||
const REASON_ROLLED_UP_DAILY := "rolled_up_daily"
|
||||
|
||||
var _last_error := ""
|
||||
var _last_telemetry: Dictionary = {}
|
||||
|
||||
|
||||
func plan(
|
||||
source: WorldEventStore,
|
||||
current_tick: int,
|
||||
ticks_per_day: int,
|
||||
raw_retention_days: int,
|
||||
high_volume_event_types: Array,
|
||||
pinned_fact_ids: Array = [],
|
||||
causal_fact_ids: Array = []
|
||||
) -> WorldEventRetentionPlan:
|
||||
_last_error = ""
|
||||
_last_telemetry = {
|
||||
"succeeded": false,
|
||||
"input_event_count": source.size() if source != null else 0,
|
||||
"error": "",
|
||||
}
|
||||
if source == null:
|
||||
return _fail("source WorldEventStore is null")
|
||||
if current_tick < 0 or ticks_per_day <= 0 or raw_retention_days < 0:
|
||||
return _fail("retention time configuration is invalid")
|
||||
var normalized_types: Variant = _normalize_event_types(high_volume_event_types)
|
||||
var normalized_pins: Variant = _normalize_event_ids(pinned_fact_ids)
|
||||
var normalized_causal_facts: Variant = _normalize_event_ids(causal_fact_ids)
|
||||
if normalized_types == null:
|
||||
return _fail("high-volume event types are invalid or duplicated")
|
||||
if normalized_pins == null:
|
||||
return _fail("pinned fact IDs are invalid or duplicated")
|
||||
if normalized_causal_facts == null:
|
||||
return _fail("causal fact IDs are invalid or duplicated")
|
||||
for event_id: int in normalized_pins:
|
||||
if not source.has_event(event_id):
|
||||
return _fail("pinned fact ID %d is missing from the source store" % event_id)
|
||||
for event_id: int in normalized_causal_facts:
|
||||
if not source.has_event(event_id):
|
||||
return _fail("causal fact ID %d is missing from the source store" % event_id)
|
||||
var source_serialized := JSON.stringify(source.to_dictionary())
|
||||
var source_checksum := source_serialized.sha256_text()
|
||||
var events := source.get_all()
|
||||
var events_by_id: Dictionary = {}
|
||||
var order_by_id: Dictionary = {}
|
||||
var traces_by_id: Dictionary = {}
|
||||
for event_index in range(events.size()):
|
||||
var event := events[event_index] as WorldEventRecord
|
||||
if event == null or event.get_tick() > current_tick:
|
||||
return _fail("source contains a null or future event")
|
||||
var event_id := event.get_event_id()
|
||||
events_by_id[event_id] = event
|
||||
order_by_id[event_id] = event_index
|
||||
var cause_ids: Variant = _extract_cause_event_ids(event.get_payload())
|
||||
if cause_ids == null:
|
||||
return _fail("event %d has an invalid causal fact contract" % event_id)
|
||||
traces_by_id[event_id] = {
|
||||
"event_id": event_id,
|
||||
"decision": "",
|
||||
"reasons": [],
|
||||
"cause_event_ids": cause_ids,
|
||||
"required_by_event_ids": [],
|
||||
"rollup_keys": [],
|
||||
}
|
||||
for event: WorldEventRecord in events:
|
||||
var event_id := event.get_event_id()
|
||||
var trace: Dictionary = traces_by_id[event_id]
|
||||
for cause_event_id: int in trace["cause_event_ids"]:
|
||||
if not events_by_id.has(cause_event_id):
|
||||
return _fail(
|
||||
"event %d references missing causal fact %d" % [event_id, cause_event_id]
|
||||
)
|
||||
if int(order_by_id[cause_event_id]) >= int(order_by_id[event_id]):
|
||||
return _fail(
|
||||
"event %d references non-prior causal fact %d" % [event_id, cause_event_id]
|
||||
)
|
||||
var high_volume_type_set: Dictionary = {}
|
||||
for event_type: StringName in normalized_types:
|
||||
high_volume_type_set[String(event_type)] = true
|
||||
var current_day_index := int(current_tick / ticks_per_day)
|
||||
var first_raw_day_index := maxi(current_day_index - raw_retention_days, 0)
|
||||
var retained: Dictionary = {}
|
||||
var contributions_by_event_id: Dictionary = {}
|
||||
for event: WorldEventRecord in events:
|
||||
var event_id := event.get_event_id()
|
||||
var event_day_index := int(event.get_tick() / ticks_per_day)
|
||||
if event_day_index >= first_raw_day_index:
|
||||
_mark_retained(retained, traces_by_id, event_id, REASON_WITHIN_RAW_RETENTION)
|
||||
continue
|
||||
if not high_volume_type_set.has(String(event.get_event_type())):
|
||||
_mark_retained(retained, traces_by_id, event_id, REASON_NOT_ROLLUP_ELIGIBLE)
|
||||
continue
|
||||
var contributions: Variant = _extract_rollup_contributions(event, event_day_index)
|
||||
if contributions == null:
|
||||
_mark_retained(retained, traces_by_id, event_id, REASON_ROLLUP_PAYLOAD_INVALID)
|
||||
continue
|
||||
contributions_by_event_id[event_id] = contributions
|
||||
for event_id: int in normalized_pins:
|
||||
_mark_retained(retained, traces_by_id, event_id, REASON_PINNED_FACT)
|
||||
for event_id: int in normalized_causal_facts:
|
||||
_mark_retained(retained, traces_by_id, event_id, REASON_EXPLICIT_CAUSAL_FACT)
|
||||
for event: WorldEventRecord in events:
|
||||
var required_by_id := event.get_event_id()
|
||||
var child_trace: Dictionary = traces_by_id[required_by_id]
|
||||
for cause_event_id: int in child_trace["cause_event_ids"]:
|
||||
_mark_retained(retained, traces_by_id, cause_event_id, REASON_CAUSAL_ANCESTOR)
|
||||
var cause_trace: Dictionary = traces_by_id[cause_event_id]
|
||||
var required_by_ids: Array = cause_trace["required_by_event_ids"]
|
||||
if required_by_id not in required_by_ids:
|
||||
required_by_ids.append(required_by_id)
|
||||
var accumulators: Dictionary = {}
|
||||
var rolled_up: Dictionary = {}
|
||||
for event: WorldEventRecord in events:
|
||||
var event_id := event.get_event_id()
|
||||
if retained.has(event_id):
|
||||
continue
|
||||
if not contributions_by_event_id.has(event_id):
|
||||
return _fail("an event selected for rollup has no validated contribution")
|
||||
var trace: Dictionary = traces_by_id[event_id]
|
||||
_add_reason(trace, REASON_ROLLED_UP_DAILY)
|
||||
for contribution: Dictionary in contributions_by_event_id[event_id]:
|
||||
var rollup_key := WorldEventDailyRollupRecord.index_key_for(
|
||||
int(contribution["day_index"]),
|
||||
StringName(contribution["world_id"]),
|
||||
StringName(contribution["location_id"]),
|
||||
StringName(contribution["event_type"]),
|
||||
StringName(contribution["item_id"])
|
||||
)
|
||||
if not accumulators.has(rollup_key):
|
||||
accumulators[rollup_key] = {
|
||||
"day_index": int(contribution["day_index"]),
|
||||
"world_id": String(contribution["world_id"]),
|
||||
"location_id": String(contribution["location_id"]),
|
||||
"event_type": String(contribution["event_type"]),
|
||||
"item_id": String(contribution["item_id"]),
|
||||
"source_event_ids": [],
|
||||
"total_amount": 0.0,
|
||||
"minimum_tick": event.get_tick(),
|
||||
"maximum_tick": event.get_tick(),
|
||||
}
|
||||
var accumulator: Dictionary = accumulators[rollup_key]
|
||||
var source_event_ids: Array = accumulator["source_event_ids"]
|
||||
source_event_ids.append(event_id)
|
||||
accumulator["total_amount"] = (
|
||||
float(accumulator["total_amount"]) + float(contribution["amount"])
|
||||
)
|
||||
accumulator["minimum_tick"] = mini(int(accumulator["minimum_tick"]), event.get_tick())
|
||||
accumulator["maximum_tick"] = maxi(int(accumulator["maximum_tick"]), event.get_tick())
|
||||
var trace_rollup_keys: Array = trace["rollup_keys"]
|
||||
trace_rollup_keys.append(rollup_key)
|
||||
rolled_up[event_id] = true
|
||||
var rollup_store := WorldEventDailyRollupStore.new()
|
||||
var accumulator_keys: Array = accumulators.keys()
|
||||
accumulator_keys.sort()
|
||||
for rollup_key: String in accumulator_keys:
|
||||
var accumulator: Dictionary = accumulators[rollup_key]
|
||||
var record := WorldEventDailyRollupRecord.create(
|
||||
int(accumulator["day_index"]),
|
||||
StringName(accumulator["world_id"]),
|
||||
StringName(accumulator["location_id"]),
|
||||
StringName(accumulator["event_type"]),
|
||||
StringName(accumulator["item_id"]),
|
||||
_copy_int_array(accumulator["source_event_ids"]),
|
||||
float(accumulator["total_amount"]),
|
||||
int(accumulator["minimum_tick"]),
|
||||
int(accumulator["maximum_tick"])
|
||||
)
|
||||
if record == null or not rollup_store.append(record):
|
||||
return _fail("daily rollup record construction failed")
|
||||
var retained_ids := _sorted_int_keys(retained)
|
||||
var rolled_up_ids := _sorted_int_keys(rolled_up)
|
||||
var trace_event_ids := _sorted_int_keys(traces_by_id)
|
||||
var reason_traces: Array[Dictionary] = []
|
||||
for event_id: int in trace_event_ids:
|
||||
var trace: Dictionary = traces_by_id[event_id]
|
||||
var reasons: Array = trace["reasons"]
|
||||
var required_by_ids: Array = trace["required_by_event_ids"]
|
||||
var rollup_keys: Array = trace["rollup_keys"]
|
||||
reasons.sort()
|
||||
required_by_ids.sort()
|
||||
rollup_keys.sort()
|
||||
trace["decision"] = (
|
||||
WorldEventRetentionPlan.DECISION_RETAIN_RAW
|
||||
if retained.has(event_id)
|
||||
else WorldEventRetentionPlan.DECISION_ROLLED_UP
|
||||
)
|
||||
reason_traces.append(trace.duplicate(true))
|
||||
var source_unchanged := JSON.stringify(source.to_dictionary()) == source_serialized
|
||||
if not source_unchanged:
|
||||
return _fail("source WorldEventStore changed while planning")
|
||||
var telemetry := {
|
||||
"succeeded": true,
|
||||
"input_event_count": events.size(),
|
||||
"retained_event_count": retained_ids.size(),
|
||||
"rolled_up_event_count": rolled_up_ids.size(),
|
||||
"rollup_record_count": rollup_store.size(),
|
||||
"rollup_contribution_count": _rollup_contribution_count(rollup_store),
|
||||
"pinned_fact_count": normalized_pins.size(),
|
||||
"causal_fact_count": normalized_causal_facts.size(),
|
||||
"causal_ancestor_count": _reason_count(reason_traces, REASON_CAUSAL_ANCESTOR),
|
||||
"recent_event_count": _reason_count(reason_traces, REASON_WITHIN_RAW_RETENTION),
|
||||
"non_rollup_event_count": _reason_count(reason_traces, REASON_NOT_ROLLUP_ELIGIBLE),
|
||||
"invalid_rollup_event_count": _reason_count(reason_traces, REASON_ROLLUP_PAYLOAD_INVALID),
|
||||
"first_raw_day_index": first_raw_day_index,
|
||||
"source_checksum": source_checksum,
|
||||
"source_unchanged": source_unchanged,
|
||||
}
|
||||
var result := WorldEventRetentionPlan.create(
|
||||
current_tick,
|
||||
ticks_per_day,
|
||||
raw_retention_days,
|
||||
first_raw_day_index,
|
||||
source_checksum,
|
||||
_copy_string_name_array(normalized_types),
|
||||
_copy_int_array(normalized_pins),
|
||||
_copy_int_array(normalized_causal_facts),
|
||||
retained_ids,
|
||||
rolled_up_ids,
|
||||
rollup_store,
|
||||
reason_traces,
|
||||
telemetry
|
||||
)
|
||||
if result == null:
|
||||
return _fail("retention plan validation failed")
|
||||
_last_telemetry = telemetry.duplicate(true)
|
||||
return result
|
||||
|
||||
|
||||
func get_last_error() -> String:
|
||||
return _last_error
|
||||
|
||||
|
||||
func get_last_telemetry() -> Dictionary:
|
||||
return _last_telemetry.duplicate(true)
|
||||
|
||||
|
||||
func _fail(message: String) -> WorldEventRetentionPlan:
|
||||
_last_error = message
|
||||
_last_telemetry["error"] = message
|
||||
_last_telemetry["succeeded"] = false
|
||||
return null
|
||||
|
||||
|
||||
static func _mark_retained(
|
||||
retained: Dictionary, traces_by_id: Dictionary, event_id: int, reason: String
|
||||
) -> void:
|
||||
retained[event_id] = true
|
||||
var trace: Dictionary = traces_by_id[event_id]
|
||||
_add_reason(trace, reason)
|
||||
|
||||
|
||||
static func _add_reason(trace: Dictionary, reason: String) -> void:
|
||||
var reasons: Array = trace["reasons"]
|
||||
if reason not in reasons:
|
||||
reasons.append(reason)
|
||||
|
||||
|
||||
static func _extract_cause_event_ids(payload: Dictionary) -> Variant:
|
||||
var seen: Dictionary = {}
|
||||
var event_ids: Array[int] = []
|
||||
if payload.has("cause_event_id"):
|
||||
var single_cause: Variant = payload["cause_event_id"]
|
||||
if not single_cause is int or int(single_cause) < -1:
|
||||
return null
|
||||
if int(single_cause) >= 0:
|
||||
seen[int(single_cause)] = true
|
||||
event_ids.append(int(single_cause))
|
||||
if payload.has("cause_event_ids"):
|
||||
var multiple_causes: Variant = payload["cause_event_ids"]
|
||||
if not multiple_causes is Array:
|
||||
return null
|
||||
for raw_cause: Variant in multiple_causes:
|
||||
if not raw_cause is int or int(raw_cause) < 0:
|
||||
return null
|
||||
if seen.has(int(raw_cause)):
|
||||
continue
|
||||
seen[int(raw_cause)] = true
|
||||
event_ids.append(int(raw_cause))
|
||||
event_ids.sort()
|
||||
return event_ids
|
||||
|
||||
|
||||
static func _extract_rollup_contributions(event: WorldEventRecord, day_index: int) -> Variant:
|
||||
if event == null or not event.has_location():
|
||||
return null
|
||||
var location := event.get_location()
|
||||
if (
|
||||
location == null
|
||||
or location.get_world_id().is_empty()
|
||||
or location.get_location_id().is_empty()
|
||||
):
|
||||
return null
|
||||
var payload := event.get_payload()
|
||||
var has_item_amount := payload.has("item_id") or payload.has("amount")
|
||||
var has_cargo_ledger := payload.has("cargo_ledger")
|
||||
if has_item_amount and has_cargo_ledger:
|
||||
return null
|
||||
var contributions: Array[Dictionary] = []
|
||||
if has_cargo_ledger:
|
||||
if not payload["cargo_ledger"] is Dictionary:
|
||||
return null
|
||||
var ledger: Dictionary = payload["cargo_ledger"]
|
||||
var source_keys: Dictionary = {}
|
||||
for raw_item_id: Variant in ledger:
|
||||
if not raw_item_id is String and not raw_item_id is StringName:
|
||||
return null
|
||||
var item_id := String(raw_item_id)
|
||||
if item_id.is_empty() or source_keys.has(item_id):
|
||||
return null
|
||||
var raw_amount: Variant = ledger[raw_item_id]
|
||||
if (
|
||||
not raw_amount is int and not raw_amount is float
|
||||
or not is_finite(float(raw_amount))
|
||||
or float(raw_amount) <= 0.0
|
||||
):
|
||||
return null
|
||||
source_keys[item_id] = raw_item_id
|
||||
var item_ids: Array = source_keys.keys()
|
||||
item_ids.sort()
|
||||
for item_id: String in item_ids:
|
||||
contributions.append(
|
||||
_contribution(
|
||||
event, location, day_index, item_id, float(ledger[source_keys[item_id]])
|
||||
)
|
||||
)
|
||||
elif payload.has("item_id") and payload.has("amount"):
|
||||
var raw_item_id: Variant = payload["item_id"]
|
||||
var raw_amount: Variant = payload["amount"]
|
||||
if (
|
||||
(not raw_item_id is String and not raw_item_id is StringName)
|
||||
or String(raw_item_id).is_empty()
|
||||
or (not raw_amount is int and not raw_amount is float)
|
||||
or not is_finite(float(raw_amount))
|
||||
or float(raw_amount) <= 0.0
|
||||
):
|
||||
return null
|
||||
contributions.append(
|
||||
_contribution(event, location, day_index, String(raw_item_id), float(raw_amount))
|
||||
)
|
||||
else:
|
||||
return null
|
||||
return contributions if not contributions.is_empty() else null
|
||||
|
||||
|
||||
static func _contribution(
|
||||
event: WorldEventRecord,
|
||||
location: SpatialAddress,
|
||||
day_index: int,
|
||||
item_id: String,
|
||||
amount: float
|
||||
) -> Dictionary:
|
||||
return {
|
||||
"day_index": day_index,
|
||||
"world_id": String(location.get_world_id()),
|
||||
"location_id": String(location.get_location_id()),
|
||||
"event_type": String(event.get_event_type()),
|
||||
"item_id": item_id,
|
||||
"amount": amount,
|
||||
}
|
||||
|
||||
|
||||
static func _normalize_event_types(values: Array) -> Variant:
|
||||
var seen: Dictionary = {}
|
||||
var normalized: Array[StringName] = []
|
||||
for value: Variant in values:
|
||||
if (not value is String and not value is StringName) or StringName(value).is_empty():
|
||||
return null
|
||||
var event_type := StringName(value)
|
||||
if seen.has(String(event_type)):
|
||||
return null
|
||||
seen[String(event_type)] = true
|
||||
normalized.append(event_type)
|
||||
normalized.sort_custom(
|
||||
func(first: StringName, second: StringName) -> bool: return String(first) < String(second)
|
||||
)
|
||||
return normalized
|
||||
|
||||
|
||||
static func _normalize_event_ids(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 _sorted_int_keys(values: Dictionary) -> Array[int]:
|
||||
var result: Array[int] = []
|
||||
for value: Variant in values:
|
||||
result.append(int(value))
|
||||
result.sort()
|
||||
return result
|
||||
|
||||
|
||||
static func _copy_int_array(values: Array) -> Array[int]:
|
||||
var result: Array[int] = []
|
||||
for value: Variant in values:
|
||||
result.append(int(value))
|
||||
return result
|
||||
|
||||
|
||||
static func _copy_string_name_array(values: Array) -> Array[StringName]:
|
||||
var result: Array[StringName] = []
|
||||
for value: Variant in values:
|
||||
result.append(StringName(value))
|
||||
return result
|
||||
|
||||
|
||||
static func _reason_count(traces: Array[Dictionary], reason: String) -> int:
|
||||
var count := 0
|
||||
for trace: Dictionary in traces:
|
||||
var reasons: Array = trace["reasons"]
|
||||
if reason in reasons:
|
||||
count += 1
|
||||
return count
|
||||
|
||||
|
||||
static func _rollup_contribution_count(store: WorldEventDailyRollupStore) -> int:
|
||||
var count := 0
|
||||
for record: WorldEventDailyRollupRecord in store.get_all():
|
||||
count += record.get_event_count()
|
||||
return count
|
||||
@@ -0,0 +1 @@
|
||||
uid://d2s2p1p8556nc
|
||||
@@ -0,0 +1,244 @@
|
||||
class_name WorldEventDailyRollupRecord
|
||||
extends RefCounted
|
||||
|
||||
const SCHEMA_VERSION := 1
|
||||
const RECORD_FIELDS := [
|
||||
"schema_version",
|
||||
"day_index",
|
||||
"world_id",
|
||||
"location_id",
|
||||
"event_type",
|
||||
"item_id",
|
||||
"event_count",
|
||||
"total_amount",
|
||||
"minimum_tick",
|
||||
"maximum_tick",
|
||||
"minimum_event_id",
|
||||
"maximum_event_id",
|
||||
"source_event_ids_checksum",
|
||||
]
|
||||
|
||||
var _day_index := -1
|
||||
var _world_id: StringName
|
||||
var _location_id: StringName
|
||||
var _event_type: StringName
|
||||
var _item_id: StringName
|
||||
var _event_count := 0
|
||||
var _total_amount := 0.0
|
||||
var _minimum_tick := -1
|
||||
var _maximum_tick := -1
|
||||
var _minimum_event_id := -1
|
||||
var _maximum_event_id := -1
|
||||
var _source_event_ids_checksum := ""
|
||||
|
||||
|
||||
static func create(
|
||||
day_index: int,
|
||||
world_id: StringName,
|
||||
location_id: StringName,
|
||||
event_type: StringName,
|
||||
item_id: StringName,
|
||||
source_event_ids: Array[int],
|
||||
total_amount: float,
|
||||
minimum_tick: int,
|
||||
maximum_tick: int
|
||||
) -> WorldEventDailyRollupRecord:
|
||||
var normalized_ids: Variant = _normalize_event_ids(source_event_ids)
|
||||
if normalized_ids == null or normalized_ids.is_empty():
|
||||
return null
|
||||
var record := WorldEventDailyRollupRecord.new()
|
||||
record._day_index = day_index
|
||||
record._world_id = world_id
|
||||
record._location_id = location_id
|
||||
record._event_type = event_type
|
||||
record._item_id = item_id
|
||||
record._event_count = normalized_ids.size()
|
||||
record._total_amount = total_amount
|
||||
record._minimum_tick = minimum_tick
|
||||
record._maximum_tick = maximum_tick
|
||||
record._minimum_event_id = int(normalized_ids.front())
|
||||
record._maximum_event_id = int(normalized_ids.back())
|
||||
record._source_event_ids_checksum = JSON.stringify(normalized_ids).sha256_text()
|
||||
return record if record.is_valid() else null
|
||||
|
||||
|
||||
static func from_dictionary(record_data: Dictionary) -> WorldEventDailyRollupRecord:
|
||||
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["day_index"] is int
|
||||
or not record_data["world_id"] is String
|
||||
or not record_data["location_id"] is String
|
||||
or not record_data["event_type"] is String
|
||||
or not record_data["item_id"] is String
|
||||
or not record_data["event_count"] is int
|
||||
or (not record_data["total_amount"] is int and not record_data["total_amount"] is float)
|
||||
or not record_data["minimum_tick"] is int
|
||||
or not record_data["maximum_tick"] is int
|
||||
or not record_data["minimum_event_id"] is int
|
||||
or not record_data["maximum_event_id"] is int
|
||||
or not record_data["source_event_ids_checksum"] is String
|
||||
):
|
||||
return null
|
||||
var record := WorldEventDailyRollupRecord.new()
|
||||
record._day_index = int(record_data["day_index"])
|
||||
record._world_id = StringName(record_data["world_id"])
|
||||
record._location_id = StringName(record_data["location_id"])
|
||||
record._event_type = StringName(record_data["event_type"])
|
||||
record._item_id = StringName(record_data["item_id"])
|
||||
record._event_count = int(record_data["event_count"])
|
||||
record._total_amount = float(record_data["total_amount"])
|
||||
record._minimum_tick = int(record_data["minimum_tick"])
|
||||
record._maximum_tick = int(record_data["maximum_tick"])
|
||||
record._minimum_event_id = int(record_data["minimum_event_id"])
|
||||
record._maximum_event_id = int(record_data["maximum_event_id"])
|
||||
record._source_event_ids_checksum = String(record_data["source_event_ids_checksum"])
|
||||
return record if record.is_valid() else null
|
||||
|
||||
|
||||
func is_valid() -> bool:
|
||||
return (
|
||||
_day_index >= 0
|
||||
and not _world_id.is_empty()
|
||||
and not _location_id.is_empty()
|
||||
and not _event_type.is_empty()
|
||||
and not _item_id.is_empty()
|
||||
and _event_count > 0
|
||||
and is_finite(_total_amount)
|
||||
and _total_amount > 0.0
|
||||
and _minimum_tick >= 0
|
||||
and _maximum_tick >= _minimum_tick
|
||||
and _minimum_event_id >= 0
|
||||
and _maximum_event_id >= _minimum_event_id
|
||||
and _is_sha256(_source_event_ids_checksum)
|
||||
)
|
||||
|
||||
|
||||
func get_day_index() -> int:
|
||||
return _day_index
|
||||
|
||||
|
||||
func get_world_id() -> StringName:
|
||||
return _world_id
|
||||
|
||||
|
||||
func get_location_id() -> StringName:
|
||||
return _location_id
|
||||
|
||||
|
||||
func get_event_type() -> StringName:
|
||||
return _event_type
|
||||
|
||||
|
||||
func get_item_id() -> StringName:
|
||||
return _item_id
|
||||
|
||||
|
||||
func get_event_count() -> int:
|
||||
return _event_count
|
||||
|
||||
|
||||
func get_total_amount() -> float:
|
||||
return _total_amount
|
||||
|
||||
|
||||
func get_minimum_tick() -> int:
|
||||
return _minimum_tick
|
||||
|
||||
|
||||
func get_maximum_tick() -> int:
|
||||
return _maximum_tick
|
||||
|
||||
|
||||
func get_minimum_event_id() -> int:
|
||||
return _minimum_event_id
|
||||
|
||||
|
||||
func get_maximum_event_id() -> int:
|
||||
return _maximum_event_id
|
||||
|
||||
|
||||
func get_source_event_ids_checksum() -> String:
|
||||
return _source_event_ids_checksum
|
||||
|
||||
|
||||
func index_key() -> String:
|
||||
return index_key_for(_day_index, _world_id, _location_id, _event_type, _item_id)
|
||||
|
||||
|
||||
func location_index_key() -> String:
|
||||
return location_key_for(_world_id, _location_id)
|
||||
|
||||
|
||||
func get_reason_trace() -> Array[StringName]:
|
||||
return [
|
||||
&"daily_transaction_rollup",
|
||||
StringName("source_event_count_%d" % _event_count),
|
||||
StringName("source_event_ids_sha256_%s" % _source_event_ids_checksum),
|
||||
]
|
||||
|
||||
|
||||
func to_dictionary() -> Dictionary:
|
||||
return {
|
||||
"schema_version": SCHEMA_VERSION,
|
||||
"day_index": _day_index,
|
||||
"world_id": String(_world_id),
|
||||
"location_id": String(_location_id),
|
||||
"event_type": String(_event_type),
|
||||
"item_id": String(_item_id),
|
||||
"event_count": _event_count,
|
||||
"total_amount": _total_amount,
|
||||
"minimum_tick": _minimum_tick,
|
||||
"maximum_tick": _maximum_tick,
|
||||
"minimum_event_id": _minimum_event_id,
|
||||
"maximum_event_id": _maximum_event_id,
|
||||
"source_event_ids_checksum": _source_event_ids_checksum,
|
||||
}
|
||||
|
||||
|
||||
static func index_key_for(
|
||||
day_index: int,
|
||||
world_id: StringName,
|
||||
location_id: StringName,
|
||||
event_type: StringName,
|
||||
item_id: StringName
|
||||
) -> String:
|
||||
return JSON.stringify(
|
||||
[day_index, String(world_id), String(location_id), String(event_type), String(item_id)]
|
||||
)
|
||||
|
||||
|
||||
static func location_key_for(world_id: StringName, location_id: StringName) -> String:
|
||||
return JSON.stringify([String(world_id), String(location_id)])
|
||||
|
||||
|
||||
static func _normalize_event_ids(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 _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
|
||||
@@ -0,0 +1 @@
|
||||
uid://b73dsvxt8phkq
|
||||
Reference in New Issue
Block a user