macula_neuroevolution_esdb

%% @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
%%   - lc-{realm}.{silo} : Liquid Conglomerate silo 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.
%% LC silo events (with 'silo' key) are routed to lc-REALM.SILO streams.
-spec event_to_stream(map()) -> binary().
event_to_stream(Event) ->
    case maps:get(silo, Event, undefined) of
        undefined ->
            %% Standard lineage event
            EventType = maps:get(event_type, Event, unknown),
            route_by_event_type(EventType, Event);
        SiloType ->
            %% LC silo event
            route_lc_event(SiloType, Event)
    end.

%%====================================================================
%% 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 LC silo events to lc-REALM.SILO streams.
route_lc_event(SiloType, Event) ->
    Realm = maps:get(realm, Event, <<"default">>),
    SiloTypeBin = atom_to_binary(SiloType, utf8),
    <<"lc-", Realm/binary, ".", SiloTypeBin/binary>>.

%% @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() ->
    <<A:32, B:16, C:16, D:16, E:48>> = 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>>.