%% @doc ESDB backend for neuroevolution lineage tracking. %% %% Implements the neuroevolution_lineage_events behaviour using %% erl-esdb-gater for event persistence. %% %% == Performance Design == %% %% This backend is designed to NEVER block the evolution loop: %% %% - persist_event/persist_batch spawn async writers %% - Returns immediately, I/O happens in background %% - Errors are logged, not propagated (fire-and-forget) %% - Under extreme load, events may be lost (acceptable) %% %% == Stream Routing == %% %% Events are routed to streams based on entity type: %% %% - individual-{id} : Birth, death, fitness, mutation events %% - species-{id} : Speciation, lineage divergence/merge %% - population-{id} : Generation, capacity, catastrophe events %% - coalition-{id} : Coalition lifecycle events %% %% @author Macula.io %% @copyright 2025 Macula.io -module(esdb_lineage_backend). -behaviour(neuroevolution_lineage_events). %% Behaviour callbacks -export([ init/1, persist_event/2, persist_batch/2, read_stream/3, subscribe/3, unsubscribe/3 ]). %% API -export([ event_to_stream/1 ]). -record(state, { store_id :: atom() }). -type state() :: #state{}. %%==================================================================== %% Behaviour Callbacks %%==================================================================== %% @doc Initialize the backend with configuration. %% %% Required config keys: %% - store_id: atom() - The erl-esdb store identifier -spec init(map()) -> {ok, state()} | {error, term()}. init(Config) -> StoreId = maps:get(store_id, Config, lineage_store), {ok, #state{store_id = StoreId}}. %% @doc Persist a single event (fire-and-forget). %% Spawns async writer and returns immediately. -spec persist_event(map(), state()) -> ok. persist_event(Event, #state{store_id = StoreId}) -> spawn_writer(StoreId, [Event]), ok. %% @doc Persist a batch of events (fire-and-forget). %% Spawns async writer and returns immediately. -spec persist_batch([map()], state()) -> ok. persist_batch([], _State) -> ok; persist_batch(Events, #state{store_id = StoreId}) -> spawn_writer(StoreId, Events), ok. %% @doc Read events from a stream. %% This MAY block - only use for recovery/replay, not during evolution. -spec read_stream(binary(), map(), state()) -> {ok, [map()]} | {error, term()}. read_stream(StreamId, Opts, #state{store_id = StoreId}) -> From = maps:get(from, Opts, 0), Limit = maps:get(limit, Opts, 1000), Direction = maps:get(direction, Opts, forward), esdb_gater_api:get_events(StoreId, StreamId, From, Limit, Direction). %% @doc Subscribe to new events on a stream. -spec subscribe(binary(), pid(), state()) -> ok | {error, term()}. subscribe(StreamId, Pid, #state{store_id = StoreId}) -> Topic = lineage_topic(StoreId, StreamId), esdb_channel_server:subscribe(esdb_channel_events, Topic, Pid). %% @doc Unsubscribe from a stream. -spec unsubscribe(binary(), pid(), state()) -> ok | {error, term()}. unsubscribe(StreamId, Pid, #state{store_id = StoreId}) -> Topic = lineage_topic(StoreId, StreamId), esdb_channel_server:unsubscribe(esdb_channel_events, Topic, Pid), ok. %%==================================================================== %% API %%==================================================================== %% @doc Route an event to its target stream based on event type. -spec event_to_stream(map()) -> binary(). event_to_stream(Event) -> EventType = maps:get(event_type, Event, unknown), route_by_event_type(EventType, Event). %%==================================================================== %% Internal Functions - Async Writer %%==================================================================== %% @private Spawn an async writer process. %% Fire-and-forget: errors are logged, not propagated. spawn_writer(StoreId, Events) -> spawn(fun() -> write_events(StoreId, Events) end). %% @private Write events to the store, grouping by stream. write_events(StoreId, Events) -> %% Group events by target stream GroupedEvents = group_by_stream(Events), %% Write each group maps:foreach( fun(StreamId, StreamEvents) -> write_stream_events(StoreId, StreamId, StreamEvents) end, GroupedEvents ). %% @private Group events by their target stream. group_by_stream(Events) -> lists:foldl( fun(Event, Acc) -> StreamId = event_to_stream(Event), EventWithMeta = add_metadata(Event), maps:update_with( StreamId, fun(Existing) -> [EventWithMeta | Existing] end, [EventWithMeta], Acc ) end, #{}, Events ). %% @private Write events to a single stream. %% Logs errors but doesn't propagate them. write_stream_events(StoreId, StreamId, Events) -> %% Reverse to maintain insertion order OrderedEvents = lists:reverse(Events), case esdb_gater_api:append_events(StoreId, StreamId, OrderedEvents) of {ok, _Version} -> ok; {error, Reason} -> error_logger:warning_msg( "[esdb_lineage_backend] Failed to write ~p events to ~s: ~p~n", [length(OrderedEvents), StreamId, Reason] ) end. %%==================================================================== %% Internal Functions - Stream Routing %%==================================================================== %% @private Route events to streams based on type route_by_event_type(EventType, Event) when EventType =:= offspring_born; EventType =:= pioneer_spawned; EventType =:= clone_produced; EventType =:= immigrant_arrived; EventType =:= individual_culled; EventType =:= lifespan_expired; EventType =:= individual_perished; EventType =:= individual_matured; EventType =:= fertility_waned; EventType =:= fitness_evaluated; EventType =:= fitness_improved; EventType =:= fitness_declined; EventType =:= champion_crowned; EventType =:= mutation_applied; EventType =:= neuron_added; EventType =:= neuron_removed; EventType =:= connection_added; EventType =:= connection_removed; EventType =:= weight_perturbed; EventType =:= knowledge_transferred; EventType =:= skill_imitated; EventType =:= behavior_cloned; EventType =:= weights_grafted; EventType =:= structure_seeded; EventType =:= mentor_assigned; EventType =:= mentorship_concluded; EventType =:= mark_acquired; EventType =:= mark_inherited; EventType =:= mark_decayed -> IndividualId = maps:get(individual_id, Event, <<"unknown">>), <<"individual-", IndividualId/binary>>; route_by_event_type(EventType, Event) when EventType =:= lineage_diverged; EventType =:= species_emerged; EventType =:= lineage_ended; EventType =:= lineage_merged -> SpeciesId = maps:get(species_id, Event, <<"unknown">>), <<"species-", SpeciesId/binary>>; route_by_event_type(EventType, Event) when EventType =:= generation_completed; EventType =:= population_initialized; EventType =:= population_terminated; EventType =:= stagnation_detected; EventType =:= breakthrough_achieved; EventType =:= carrying_capacity_reached; EventType =:= catastrophe_occurred -> PopulationId = maps:get(population_id, Event, <<"unknown">>), <<"population-", PopulationId/binary>>; route_by_event_type(EventType, Event) when EventType =:= coalition_formed; EventType =:= coalition_dissolved; EventType =:= coalition_joined -> CoalitionId = maps:get(coalition_id, Event, <<"unknown">>), <<"coalition-", CoalitionId/binary>>; route_by_event_type(_EventType, Event) -> %% Fallback: use individual_id or population_id if present case maps:get(individual_id, Event, undefined) of undefined -> case maps:get(population_id, Event, undefined) of undefined -> <<"unknown-events">>; PopId -> <<"population-", PopId/binary>> end; IndId -> <<"individual-", IndId/binary>> end. %%==================================================================== %% Internal Functions - Utilities %%==================================================================== %% @private Add metadata to event add_metadata(Event) -> Timestamp = maps:get(timestamp, Event, erlang:system_time(millisecond)), EventId = maps:get(event_id, Event, generate_event_id()), Event#{ timestamp => Timestamp, event_id => EventId }. %% @private Generate a unique event ID generate_event_id() -> <> = crypto:strong_rand_bytes(16), iolist_to_binary(io_lib:format( "~8.16.0b-~4.16.0b-~4.16.0b-~4.16.0b-~12.16.0b", [A, B, C, D, E] )). %% @private Build lineage topic for pub/sub lineage_topic(StoreId, StreamId) -> StoreIdBin = atom_to_binary(StoreId, utf8), <<"lineage.", StoreIdBin/binary, ".", StreamId/binary>>.