TheTree_mock/scripts/core/domain_event_stream.gd

class_name DomainEventStream
extends RefCounted

const SNAPSHOT_VERSION := 1
const WILDCARD_EVENT_TYPE: StringName = &"*"
const RECENT_EVENT_LIMIT := 256
const MAX_BATCH_SIZE := 4_096

var _subscribers: Dictionary = {}
var _published_ids: Dictionary = {}
var _recent_events: Array[SimulationDomainEvent] = []
var _pending_deliveries: Array[SimulationDomainEvent] = []
var _delivery_active := false

func subscribe(event_type: StringName, subscriber_id: StringName, callback: Callable) -> bool:
	if event_type.is_empty() or subscriber_id.is_empty() or not callback.is_valid():
		return false
	if not _subscribers.has(event_type):
		_subscribers[event_type] = {}
	var subscribers: Dictionary = _subscribers[event_type]
	if subscribers.has(subscriber_id):
		return false
	subscribers[subscriber_id] = callback
	return true

func unsubscribe(event_type: StringName, subscriber_id: StringName) -> void:
	if _subscribers.has(event_type):
		var subscribers: Dictionary = _subscribers[event_type]
		subscribers.erase(subscriber_id)

func publish(event: SimulationDomainEvent) -> bool:
	return publish_batch([event])


## Validate and commit the entire event set before the first subscriber runs.
## Reentrant publications join the end of the delivery queue, so callbacks can
## never observe or deliver only a prefix of the committed batch.
func publish_batch(events: Array) -> bool:
	if events.size() > MAX_BATCH_SIZE:
		return false
	var committed: Array[SimulationDomainEvent] = []
	var batch_ids: Dictionary = {}
	for event_value: Variant in events:
		if not event_value is SimulationDomainEvent:
			return false
		var event: SimulationDomainEvent = event_value
		if event == null or not event.is_valid() \
		or _published_ids.has(event.event_id) or batch_ids.has(event.event_id):
			return false
		batch_ids[event.event_id] = true
		committed.append(event.copy())
	for event: SimulationDomainEvent in committed:
		_published_ids[event.event_id] = true
		_recent_events.append(event.copy())
		if _recent_events.size() > RECENT_EVENT_LIMIT:
			_recent_events.pop_front()
		_pending_deliveries.append(event.copy())
	_drain_deliveries()
	return true


func _drain_deliveries() -> void:
	if _delivery_active:
		return
	_delivery_active = true
	while not _pending_deliveries.is_empty():
		var event: SimulationDomainEvent = _pending_deliveries.pop_front()
		_deliver(event)
	_delivery_active = false


func _deliver(event: SimulationDomainEvent) -> void:
	var callbacks := {}
	_collect_callbacks(event.event_type, callbacks)
	_collect_callbacks(WILDCARD_EVENT_TYPE, callbacks)
	var subscriber_ids: Array = callbacks.keys()
	subscriber_ids.sort()
	for subscriber_id: Variant in subscriber_ids:
		var callback: Callable = callbacks[subscriber_id]
		if callback.is_valid():
			callback.call(event.copy())

func has_published(event_id: StringName) -> bool:
	return _published_ids.has(event_id)

func recent_events() -> Array[SimulationDomainEvent]:
	var result: Array[SimulationDomainEvent] = []
	for event in _recent_events:
		result.append(event.copy())
	return result

func create_snapshot() -> Dictionary:
	var published_ids: Array = _published_ids.keys()
	published_ids.sort_custom(func(left: Variant, right: Variant) -> bool:
		return String(left) < String(right)
	)
	var saved_ids: Array[String] = []
	for event_id: Variant in published_ids:
		saved_ids.append(String(event_id))
	var saved_recent: Array[Dictionary] = []
	for event in _recent_events:
		saved_recent.append(event.to_dictionary())
	return {
		"version": SNAPSHOT_VERSION,
		"published_ids": saved_ids,
		"recent_events": saved_recent,
	}

func restore_snapshot(snapshot: Dictionary) -> bool:
	if _delivery_active or not _pending_deliveries.is_empty():
		return false
	if int(snapshot.get("version", -1)) != SNAPSHOT_VERSION:
		return false
	if not snapshot.get("published_ids", null) is Array \
			or not snapshot.get("recent_events", null) is Array:
		return false
	var restored_ids := {}
	for event_id: Variant in snapshot["published_ids"]:
		var normalized_id := StringName(event_id)
		if normalized_id.is_empty() or restored_ids.has(normalized_id):
			return false
		restored_ids[normalized_id] = true
	var restored_recent: Array[SimulationDomainEvent] = []
	for event_data: Variant in snapshot["recent_events"]:
		if not event_data is Dictionary:
			return false
		var event := SimulationDomainEvent.from_dictionary(event_data)
		if not event.is_valid() or not restored_ids.has(event.event_id):
			return false
		restored_recent.append(event)
	if restored_recent.size() > RECENT_EVENT_LIMIT:
		return false
	_published_ids = restored_ids
	_recent_events = restored_recent
	_pending_deliveries.clear()
	return true

func _collect_callbacks(event_type: StringName, target: Dictionary) -> void:
	if not _subscribers.has(event_type):
		return
	var source: Dictionary = _subscribers[event_type]
	for subscriber_id: Variant in source:
		if not target.has(subscriber_id):
			target[subscriber_id] = source[subscriber_id]