Packages
eventsourcing_glyn
1.0.0
A distributed event sourcing library that enhances existing event stores with Glyn PubSub and Registry capabilities for distributed query actors, event broadcasting, and fault-tolerant command routing
Current section
Files
Jump to
Current section
Files
src/eventsourcing_glyn.erl
-module(eventsourcing_glyn).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch, inline]).
-define(FILEPATH, "src/eventsourcing_glyn.gleam").
-export([get_glyn_store/1, memory_store_event_envelop_decoder/1, metadata_decoder/0, serialized_event_envelop_decoder/1, payload_decoder/1, supervised/4, subscribers/1, publish/3]).
-export_type([glyn_store/6, glyn_config/0, glyn_instances/1]).
-if(?OTP_RELEASE >= 27).
-define(MODULEDOC(Str), -moduledoc(Str)).
-define(DOC(Str), -doc(Str)).
-else.
-define(MODULEDOC(Str), -compile([])).
-define(DOC(Str), -compile([])).
-endif.
-opaque glyn_store(IMP, IMQ, IMR, IMS, IMT, IMU) :: {glyn_store,
binary(),
binary(),
glyn_instances(IMS),
eventsourcing:event_store(IMP, IMQ, IMR, IMS, IMT, IMU)}.
-type glyn_config() :: {glyn_config, binary(), binary()}.
-type glyn_instances(IMV) :: {glyn_instances,
glyn@pubsub:pub_sub(eventsourcing:query_message(IMV))}.
-file("src/eventsourcing_glyn.gleam", 256).
?DOC(false).
-spec get_glyn_store(
eventsourcing:event_store(glyn_store(IOC, IOD, IOE, IOF, IOG, IOH), IOD, IOE, IOF, IOG, IOH)
) -> glyn_store(IOC, IOD, IOE, IOF, IOG, IOH).
get_glyn_store(Eventstore) ->
erlang:element(10, Eventstore).
-file("src/eventsourcing_glyn.gleam", 306).
?DOC(false).
-spec memory_store_event_envelop_decoder(gleam@dynamic@decode:decoder(IPE)) -> gleam@dynamic@decode:decoder(eventsourcing:event_envelop(IPE)).
memory_store_event_envelop_decoder(Event_decoder) ->
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Aggregate_id) ->
gleam@dynamic@decode:field(
2,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Sequence) ->
gleam@dynamic@decode:field(
3,
Event_decoder,
fun(Payload) ->
gleam@dynamic@decode:success(
{memory_store_event_envelop,
Aggregate_id,
Sequence,
Payload,
[]}
)
end
)
end
)
end
).
-file("src/eventsourcing_glyn.gleam", 332).
?DOC(false).
-spec metadata_decoder() -> gleam@dynamic@decode:decoder({binary(), binary()}).
metadata_decoder() ->
gleam@dynamic@decode:field(
0,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Key) ->
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Value) -> gleam@dynamic@decode:success({Key, Value}) end
)
end
).
-file("src/eventsourcing_glyn.gleam", 284).
?DOC(false).
-spec serialized_event_envelop_decoder(gleam@dynamic@decode:decoder(IPA)) -> gleam@dynamic@decode:decoder(eventsourcing:event_envelop(IPA)).
serialized_event_envelop_decoder(Event_decoder) ->
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Aggregate_id) ->
gleam@dynamic@decode:field(
2,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Sequence) ->
gleam@dynamic@decode:field(
3,
Event_decoder,
fun(Payload) ->
gleam@dynamic@decode:field(
4,
gleam@dynamic@decode:list(metadata_decoder()),
fun(Metadata) ->
gleam@dynamic@decode:field(
5,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Event_type) ->
gleam@dynamic@decode:field(
6,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Event_version) ->
gleam@dynamic@decode:field(
7,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Aggregate_type) ->
gleam@dynamic@decode:success(
{serialized_event_envelop,
Aggregate_id,
Sequence,
Payload,
Metadata,
Event_type,
Event_version,
Aggregate_type}
)
end
)
end
)
end
)
end
)
end
)
end
)
end
).
-file("src/eventsourcing_glyn.gleam", 323).
?DOC(false).
-spec payload_decoder(gleam@dynamic@decode:decoder(IPI)) -> gleam@dynamic@decode:decoder(eventsourcing:event_envelop(IPI)).
payload_decoder(Event_decoder) ->
gleam@dynamic@decode:one_of(
memory_store_event_envelop_decoder(Event_decoder),
[serialized_event_envelop_decoder(Event_decoder)]
).
-file("src/eventsourcing_glyn.gleam", 367).
?DOC(" Publish an event to Glyn PubSub for distribution to all query subscribers\n").
-spec publish_to_glyn_pubsub(
glyn_store(any(), any(), any(), IQN, any(), any()),
binary(),
list(eventsourcing:event_envelop(IQN))
) -> {ok, integer()} | {error, glyn@pubsub:pub_sub_error()}.
publish_to_glyn_pubsub(Glyn_instances, Aggregate_id, Events) ->
Message = {process_events, Aggregate_id, Events},
glyn@pubsub:publish(
erlang:element(2, erlang:element(4, Glyn_instances)),
erlang:element(3, Glyn_instances),
Message
).
-file("src/eventsourcing_glyn.gleam", 338).
-spec glyn_commit_events(
glyn_store(any(), IPO, IPP, IPQ, IPR, IPS),
IPS,
eventsourcing:aggregate(IPO, IPP, IPQ, IPR),
list(IPQ),
list({binary(), binary()})
) -> {ok, {list(eventsourcing:event_envelop(IPQ)), integer()}} |
{error, eventsourcing:event_sourcing_error(IPR)}.
glyn_commit_events(Glyn_store, Tx, Aggregate, Events, Metadata) ->
begin
Result = (erlang:element(6, erlang:element(5, Glyn_store)))(
Tx,
Aggregate,
Events,
Metadata
),
case Result of
{ok, X} ->
{Events@1, Version} = X,
begin
Result@1 = begin
_pipe = publish_to_glyn_pubsub(
Glyn_store,
erlang:element(2, Aggregate),
Events@1
),
case _pipe of
{ok, X@1} ->
{ok, X@1};
{error, Error} ->
{error,
begin
_@1 = Error,
{event_store_error,
<<"Failed to publish to Glyn PubSub"/utf8>>}
end}
end
end,
case Result@1 of
{ok, X@2} ->
_@2 = X@2,
{ok, {Events@1, Version}};
{error, E} ->
{error, E}
end
end;
{error, E@1} ->
{error, E@1}
end
end.
-file("src/eventsourcing_glyn.gleam", 388).
-spec proxy_load_events(
glyn_store(any(), any(), any(), IRD, IRE, any()),
binary(),
integer()
) -> {ok, list(eventsourcing:event_envelop(IRD))} |
{error, eventsourcing:event_sourcing_error(IRE)}.
proxy_load_events(Glyn_store, Aggregate_id, Start_from) ->
(erlang:element(4, erlang:element(5, Glyn_store)))(
fun(Underlying_tx) ->
(erlang:element(7, erlang:element(5, Glyn_store)))(
erlang:element(10, erlang:element(5, Glyn_store)),
Underlying_tx,
Aggregate_id,
Start_from
)
end
).
-file("src/eventsourcing_glyn.gleam", 412).
-spec proxy_load_snapshot(
glyn_store(any(), IRS, any(), any(), IRV, any()),
binary()
) -> {ok, gleam@option:option(eventsourcing:snapshot(IRS))} |
{error, eventsourcing:event_sourcing_error(IRV)}.
proxy_load_snapshot(Glyn_store, Aggregate_id) ->
(erlang:element(5, erlang:element(5, Glyn_store)))(
fun(Underlying_tx) ->
(erlang:element(8, erlang:element(5, Glyn_store)))(
Underlying_tx,
Aggregate_id
)
end
).
-file("src/eventsourcing_glyn.gleam", 428).
-spec proxy_save_snapshot(
glyn_store(any(), ISJ, any(), any(), ISM, any()),
eventsourcing:snapshot(ISJ)
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(ISM)}.
proxy_save_snapshot(Glyn_store, Snapshot) ->
(erlang:element(2, erlang:element(5, Glyn_store)))(
fun(Underlying_tx) ->
(erlang:element(9, erlang:element(5, Glyn_store)))(
Underlying_tx,
Snapshot
)
end
).
-file("src/eventsourcing_glyn.gleam", 444).
-spec proxy_execute_transaction(
glyn_store(any(), any(), any(), any(), ITC, ITD),
fun((ITD) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(ITC)})
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(ITC)}.
proxy_execute_transaction(Glyn_store, Transaction_fn) ->
(erlang:element(2, erlang:element(5, Glyn_store)))(Transaction_fn).
-file("src/eventsourcing_glyn.gleam", 458).
-spec proxy_get_latest_snapshot_transaction(
glyn_store(any(), ITP, any(), any(), ITS, ITT),
fun((ITT) -> {ok, gleam@option:option(eventsourcing:snapshot(ITP))} |
{error, eventsourcing:event_sourcing_error(ITS)})
) -> {ok, gleam@option:option(eventsourcing:snapshot(ITP))} |
{error, eventsourcing:event_sourcing_error(ITS)}.
proxy_get_latest_snapshot_transaction(Glyn_store, Transaction_fn) ->
(erlang:element(5, erlang:element(5, Glyn_store)))(Transaction_fn).
-file("src/eventsourcing_glyn.gleam", 472).
-spec proxy_load_aggregate_transaction(
glyn_store(any(), IUH, IUI, IUJ, IUK, IUL),
fun((IUL) -> {ok, eventsourcing:aggregate(IUH, IUI, IUJ, IUK)} |
{error, eventsourcing:event_sourcing_error(IUK)})
) -> {ok, eventsourcing:aggregate(IUH, IUI, IUJ, IUK)} |
{error, eventsourcing:event_sourcing_error(IUK)}.
proxy_load_aggregate_transaction(Glyn_store, Transaction_fn) ->
(erlang:element(3, erlang:element(5, Glyn_store)))(Transaction_fn).
-file("src/eventsourcing_glyn.gleam", 486).
-spec proxy_load_events_transaction(
glyn_store(any(), any(), any(), IVD, IVE, IVF),
fun((IVF) -> {ok, list(eventsourcing:event_envelop(IVD))} |
{error, eventsourcing:event_sourcing_error(IVE)})
) -> {ok, list(eventsourcing:event_envelop(IVD))} |
{error, eventsourcing:event_sourcing_error(IVE)}.
proxy_load_events_transaction(Glyn_store, Transaction_fn) ->
(erlang:element(4, erlang:element(5, Glyn_store)))(Transaction_fn).
-file("src/eventsourcing_glyn.gleam", 500).
-spec start_glyn_query_actor(
glyn_instances(IVS),
gleam@erlang@process:name(eventsourcing:query_message(IVS)),
fun((binary(), list(eventsourcing:event_envelop(IVS))) -> nil),
binary()
) -> {ok,
gleam@otp@actor:started(gleam@erlang@process:subject(eventsourcing:query_message(IVS)))} |
{error, gleam@otp@actor:start_error()}.
start_glyn_query_actor(Glyn_instances, Query_name, Query_fn, Pubsub_group) ->
begin
Result = begin
_pipe@4 = gleam@otp@actor:new_with_initialiser(
500,
fun(Subject) ->
Subscription_selector = glyn@pubsub:subscribe(
erlang:element(2, Glyn_instances),
Pubsub_group
),
Selector = begin
_pipe = gleam_erlang_ffi:new_selector(),
gleam_erlang_ffi:merge_selector(
_pipe,
Subscription_selector
)
end,
_pipe@1 = gleam@otp@actor:initialised(Subject),
_pipe@2 = gleam@otp@actor:selecting(_pipe@1, Selector),
_pipe@3 = gleam@otp@actor:returning(_pipe@2, Subject),
{ok, _pipe@3}
end
),
_pipe@5 = gleam@otp@actor:on_message(
_pipe@4,
fun(State, Message) -> case Message of
{process_events, Aggregate_id, Events} ->
gleam_stdlib:println(
<<<<<<"📥 Query actor received "/utf8,
(erlang:integer_to_binary(
erlang:length(Events)
))/binary>>/binary,
" events for aggregate: "/utf8>>/binary,
Aggregate_id/binary>>
),
Query_fn(Aggregate_id, Events),
gleam@otp@actor:continue(State)
end end
),
_pipe@6 = gleam@otp@actor:named(_pipe@5, Query_name),
gleam@otp@actor:start(_pipe@6)
end,
case Result of
{ok, X} ->
{ok, X};
{error, E} ->
{error, E}
end
end.
-file("src/eventsourcing_glyn.gleam", 149).
?DOC(
" Creates a supervised Glyn-enhanced event store that wraps an existing event store\n"
" and adds distributed PubSub capabilities for query actors and event broadcasting.\n"
" \n"
" This function creates a complete supervision tree that includes:\n"
" - Query actors that automatically subscribe to Glyn PubSub for event notifications\n"
" - The enhanced event store that publishes committed events to all subscribers\n"
" - Integration with Glyn's distributed coordination system\n"
" - Fault-tolerant supervision of all distributed components\n"
"\n"
" ## Example\n"
" ```gleam\n"
" import eventsourcing_glyn\n"
" import eventsourcing/memory_store\n"
" import gleam/otp/static_supervisor\n"
" \n"
" // 1. Set up underlying store\n"
" let #(underlying_store, memory_spec) = memory_store.supervised(\n"
" process.new_name(\"events\"),\n"
" process.new_name(\"snapshots\"), \n"
" static_supervisor.OneForOne\n"
" )\n"
" \n"
" // 2. Configure distributed capabilities\n"
" let config = eventsourcing_glyn.GlynConfig(\n"
" pubsub_scope: \"bank_events\",\n"
" pubsub_group: \"bank_services\"\n"
" )\n"
" \n"
" // 3. Define query actors\n"
" let queries = [\n"
" #(process.new_name(\"balance_query\"), balance_query_fn),\n"
" #(process.new_name(\"audit_query\"), audit_query_fn),\n"
" ]\n"
" \n"
" // 4. Create distributed event store\n"
" let assert Ok(#(glyn_eventstore, glyn_spec)) = eventsourcing_glyn.supervised(\n"
" config,\n"
" underlying_store,\n"
" queries,\n"
" my_event_decoder()\n"
" )\n"
" \n"
" // 5. Start supervision tree\n"
" let assert Ok(_) =\n"
" static_supervisor.new(static_supervisor.OneForOne)\n"
" |> static_supervisor.add(memory_spec)\n"
" |> static_supervisor.add(glyn_spec)\n"
" |> static_supervisor.start()\n"
" ```\n"
"\n"
" ## Query Actor Behavior\n"
" Each query actor receives a `QueryMessage(event)` containing:\n"
" - `ProcessEvents(aggregate_id, events)`: List of events to process\n"
" \n"
" Query functions have the signature:\n"
" ```gleam\n"
" fn(aggregate_id: String, events: List(EventEnvelop(event))) -> Nil\n"
" ```\n"
"\n"
" ## Distributed Event Flow\n"
" 1. Service A commits events via `eventsourcing.execute()`\n"
" 2. Events are persisted to the underlying store\n"
" 3. Events are automatically broadcast via Glyn PubSub\n"
" 4. All query actors across all nodes receive and process the events\n"
" 5. Query actors update their projections/views accordingly\n"
).
-spec supervised(
glyn_config(),
eventsourcing:event_store(IMW, IMX, IMY, IMZ, INA, INB),
list({gleam@erlang@process:name(eventsourcing:query_message(IMZ)),
fun((binary(), list(eventsourcing:event_envelop(IMZ))) -> nil)}),
gleam@dynamic@decode:decoder(IMZ)
) -> {ok,
{eventsourcing:event_store(glyn_store(IMW, IMX, IMY, IMZ, INA, INB), IMX, IMY, IMZ, INA, INB),
gleam@otp@supervision:child_specification(gleam@otp@static_supervisor:supervisor())}} |
{error, binary()}.
supervised(Config, Underlying_store, Queries, Event_decoder) ->
Message_decoder = begin
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Aggregate_id) ->
gleam@dynamic@decode:field(
2,
gleam@dynamic@decode:list(payload_decoder(Event_decoder)),
fun(Events) ->
gleam@dynamic@decode:success(
{process_events, Aggregate_id, Events}
)
end
)
end
)
end,
Glyn_pubsub = glyn@pubsub:new(
erlang:element(2, Config),
Message_decoder,
{process_events, <<"error"/utf8>>, []}
),
Glyn_instances = {glyn_instances, Glyn_pubsub},
Glyn_store = {glyn_store,
erlang:element(2, Config),
erlang:element(3, Config),
Glyn_instances,
Underlying_store},
Query_specs = gleam@list:map(
Queries,
fun(Query_def) ->
{Query_name, Query_fn} = Query_def,
gleam@otp@supervision:worker(
fun() ->
start_glyn_query_actor(
Glyn_instances,
Query_name,
Query_fn,
erlang:element(3, Config)
)
end
)
end
),
Eventstore = {event_store,
fun(F) -> proxy_execute_transaction(Glyn_store, F) end,
fun(F@1) -> proxy_load_aggregate_transaction(Glyn_store, F@1) end,
fun(F@2) -> proxy_load_events_transaction(Glyn_store, F@2) end,
fun(F@3) -> proxy_get_latest_snapshot_transaction(Glyn_store, F@3) end,
fun(Tx, Aggregate, Events@1, Metadata) ->
glyn_commit_events(Glyn_store, Tx, Aggregate, Events@1, Metadata)
end,
fun(Glyn_store@1, _, Aggregate_id@1, Start_from) ->
proxy_load_events(Glyn_store@1, Aggregate_id@1, Start_from)
end,
fun(_, Aggregate_id@2) ->
proxy_load_snapshot(Glyn_store, Aggregate_id@2)
end,
fun(_, Snapshot) -> proxy_save_snapshot(Glyn_store, Snapshot) end,
Glyn_store},
Supervisor_spec = begin
_pipe = gleam@otp@static_supervisor:new(one_for_one),
_pipe@1 = gleam@list:fold(
Query_specs,
_pipe,
fun(Sup, Spec) -> gleam@otp@static_supervisor:add(Sup, Spec) end
),
gleam@otp@static_supervisor:supervised(_pipe@1)
end,
{ok, {Eventstore, Supervisor_spec}}.
-file("src/eventsourcing_glyn.gleam", 557).
?DOC(
" Get the list of subscribers to this Glyn event store's PubSub scope.\n"
"\n"
" This function returns information about all query actors currently subscribed\n"
" to receive events from this distributed event store across all connected nodes.\n"
"\n"
" ## Example\n"
" ```gleam\n"
" let subscriber_info = eventsourcing_glyn.subscribers(glyn_eventstore)\n"
" // Use for monitoring or debugging distributed event flow\n"
" ```\n"
"\n"
" ## Use Cases\n"
" - **Health Monitoring**: Check if expected query actors are connected\n"
" - **Debugging**: Verify event distribution is working correctly \n"
" - **Load Balancing**: Understand the distribution of query actors\n"
" - **Troubleshooting**: Identify disconnected or failed nodes\n"
).
-spec subscribers(glyn_store(any(), any(), any(), any(), any(), any())) -> list(gleam@erlang@process:pid_()).
subscribers(Glyn_store) ->
glyn@pubsub:subscribers(
erlang:element(2, erlang:element(4, Glyn_store)),
erlang:element(2, Glyn_store)
).
-file("src/eventsourcing_glyn.gleam", 596).
?DOC(
" Manually publish an event message to distributed query subscribers.\n"
"\n"
" This function allows you to send custom messages to query actors without\n"
" going through the normal event store commit process. This is useful for:\n"
" - Broadcasting system events or notifications\n"
" - Triggering query actor maintenance or refresh operations\n"
" - Testing distributed event handling\n"
"\n"
" ## Example\n"
" ```gleam\n"
" import eventsourcing_glyn\n"
" import eventsourcing.{ProcessEvents}\n"
" \n"
" // Send a custom event to all query actors\n"
" let message = ProcessEvents(\"system\", [maintenance_event])\n"
" case eventsourcing_glyn.publish(glyn_store, \"bank_services\", message) {\n"
" Ok(count) -> io.println(\"Notified \" <> int.to_string(count) <> \" subscribers\")\n"
" Error(error) -> io.println(\"Failed to publish: \" <> string.inspect(error))\n"
" }\n"
" ```\n"
"\n"
" ## Important Notes\n"
" - **Normal Usage**: Most events should go through `eventsourcing.execute()` for proper persistence\n"
" - **Manual Publishing**: Only use this for system events or testing scenarios\n"
" - **Group Matching**: The group parameter should typically match your `GlynConfig.pubsub_group`\n"
" - **Return Value**: The count indicates how many query actors received the message\n"
).
-spec publish(
glyn_store(any(), any(), any(), IWQ, IWR, any()),
binary(),
eventsourcing:query_message(IWQ)
) -> {ok, integer()} | {error, eventsourcing:event_sourcing_error(IWR)}.
publish(Glyn_store, Group, Message) ->
_pipe = glyn@pubsub:publish(
erlang:element(2, erlang:element(4, Glyn_store)),
Group,
Message
),
case _pipe of
{ok, X} ->
{ok, X};
{error, Error} ->
{error, case Error of
{publish_failed, Error@1} ->
{event_store_error,
<<"Failed to publish to Glyn PubSub: "/utf8,
Error@1/binary>>}
end}
end.