Current section
Files
Jump to
Current section
Files
src/eventsourcing_sqlite.erl
-module(eventsourcing_sqlite).
-compile([no_auto_import, nowarn_unused_vars, nowarn_unused_function, nowarn_nomatch]).
-export([load_events/4, create_event_table/1, create_snapshot_table/1, new/8]).
-export_type([sqlite_store/4]).
-type sqlite_store(SHY, SHZ, SIA, SIB) :: {sqlite_store,
binary(),
fun((SIA) -> binary()),
gleam@dynamic@decode:decoder(SIA),
binary(),
binary(),
binary(),
fun((SHY) -> binary()),
gleam@dynamic@decode:decoder(SHY)} |
{gleam_phantom, SHZ, SIB}.
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 265).
-spec wrap_events(
sqlite_store(any(), any(), SKT, any()),
binary(),
list(SKT),
integer(),
list({binary(), binary()})
) -> list(eventsourcing:event_envelop(SKT)).
wrap_events(Postgres_store, Aggregate_id, Events, Sequence, Metadata) ->
_pipe = gleam@list:map_fold(
Events,
Sequence,
fun(Sequence@1, Event) ->
Next_sequence = Sequence@1 + 1,
{Next_sequence,
{serialized_event_envelop,
Aggregate_id,
Sequence@1 + 1,
Event,
Metadata,
erlang:element(5, Postgres_store),
erlang:element(6, Postgres_store),
erlang:element(7, Postgres_store)}}
end
),
gleam@pair:second(_pipe).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 362).
-spec metadata_encoder(list({binary(), binary()})) -> binary().
metadata_encoder(Metadata) ->
_pipe = gleam@json:array(
Metadata,
fun(Row) ->
gleam@json:preprocessed_array(
[gleam@json:string(erlang:element(1, Row)),
gleam@json:string(erlang:element(2, Row))]
)
end
),
gleam@json:to_string(_pipe).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 369).
-spec metadata_decoder() -> gleam@dynamic@decode:decoder(list({binary(),
binary()})).
metadata_decoder() ->
gleam@dynamic@decode:then(
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Stringmetadata) ->
_assert_subject = gleam@json:parse(
Stringmetadata,
gleam@dynamic@decode:list(
gleam@dynamic@decode:list(
{decoder, fun gleam@dynamic@decode:decode_string/1}
)
)
),
{ok, Listmetadata} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"metadata_decoder"/utf8>>,
line => 371})
end,
_pipe = gleam@list:map(
Listmetadata,
fun(Metadata) ->
[Key, Val] = case Metadata of
[_, _] -> Metadata;
_assert_fail@1 ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail@1,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"metadata_decoder"/utf8>>,
line => 375})
end,
{Key, Val}
end
),
gleam@dynamic@decode:success(_pipe)
end
).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 449).
-spec execute_in_transaction(binary()) -> fun((fun((sqlight:connection()) -> {ok,
SOF} |
{error, any()})) -> {ok, SOF} |
{error, eventsourcing:event_sourcing_error(any())}).
execute_in_transaction(Connection_string) ->
fun(F) ->
F@1 = fun(Db) -> _pipe = F(Db),
gleam@result:map_error(
_pipe,
fun(Error) -> gleam@string:inspect(Error) end
) end,
_pipe@1 = sqlight:with_connection(Connection_string, F@1),
gleam@result:map_error(
_pipe@1,
fun(Error@1) ->
{event_store_error, gleam@string:inspect(Error@1)}
end
)
end.
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 294).
-spec persist_events(
sqlite_store(any(), any(), SLE, any()),
sqlight:connection(),
list(eventsourcing:event_envelop(SLE))
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(any())}.
persist_events(Sqlite_store, Tx, Wrapped_events) ->
{Placeholders@1, Params@1, _} = gleam@list:index_fold(
Wrapped_events,
{<<""/utf8>>, [], 0},
fun(Acc, Event, Index) ->
{Placeholders, Params, _} = Acc,
Offset = Index * 7,
Row_placeholders = <<<<"($"/utf8,
(gleam@string:join(
begin
_pipe = gleam@list:range(Offset + 1, Offset + 7),
gleam@list:map(
_pipe,
fun erlang:integer_to_binary/1
)
end,
<<", $"/utf8>>
))/binary>>/binary,
")"/utf8>>,
Sep = case Placeholders of
<<""/utf8>> ->
<<""/utf8>>;
_ ->
<<", "/utf8>>
end,
{serialized_event_envelop,
Aggregate_id,
Sequence,
Payload,
Metadata,
Event_type,
Event_version,
Aggregate_type} = case Event of
{serialized_event_envelop, _, _, _, _, _, _, _} -> Event;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"persist_events"/utf8>>,
line => 319})
end,
New_params = [sqlight:text(Aggregate_type),
sqlight:text(Aggregate_id),
sqlight:int(Sequence),
sqlight:text(Event_type),
sqlight:text(Event_version),
sqlight:text(
begin
_pipe@1 = Payload,
(erlang:element(3, Sqlite_store))(_pipe@1)
end
),
sqlight:text(
begin
_pipe@2 = Metadata,
metadata_encoder(_pipe@2)
end
)],
{<<<<Placeholders/binary, Sep/binary>>/binary,
Row_placeholders/binary>>,
lists:append(Params, New_params),
Index + 1}
end
),
case Wrapped_events of
[] ->
{ok, nil};
_ ->
Query = <<"
INSERT INTO event
(aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata)
VALUES
"/utf8,
Placeholders@1/binary>>,
_pipe@3 = sqlight:'query'(
Query,
Tx,
Params@1,
{decoder, fun gleam@dynamic@decode:decode_dynamic/1}
),
_pipe@4 = gleam@result:map(_pipe@3, fun(_) -> nil end),
gleam@result:map_error(
_pipe@4,
fun(Error) ->
{event_store_error,
<<"Failed to insert events: "/utf8,
(gleam@string:inspect(Error))/binary>>}
end
)
end.
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 231).
-spec commit_events(
sqlite_store(SJX, SJY, SJZ, SKA),
sqlight:connection(),
eventsourcing:aggregate(SJX, SJY, SJZ, SKA),
list(SJZ),
list({binary(), binary()})
) -> {ok, {list(eventsourcing:event_envelop(SJZ)), integer()}} |
{error, eventsourcing:event_sourcing_error(SKA)}.
commit_events(Sqlite_store, Tx, Aggregate, Events, Metadata) ->
{aggregate, Aggregate_id, _, Sequence} = Aggregate,
Wrapped_events = wrap_events(
Sqlite_store,
Aggregate_id,
Events,
Sequence,
Metadata
),
_assert_subject = gleam@list:last(Wrapped_events),
{ok, Last_event} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"commit_events"/utf8>>,
line => 245})
end,
Events@1 = begin
_pipe = persist_events(Sqlite_store, Tx, Wrapped_events),
gleam@result:map(
_pipe,
fun(_) -> {Wrapped_events, erlang:element(3, Last_event)} end
)
end,
gleam@result:'try'(
begin
_pipe@1 = sqlight:exec(<<"COMMIT;"/utf8>>, Tx),
gleam@result:map_error(
_pipe@1,
fun(Error) ->
{event_store_error, gleam@string:inspect(Error)}
end
)
end,
fun(_) -> _pipe@2 = Events@1,
gleam@result:map_error(
_pipe@2,
fun(Error@1) ->
sqlight:exec(<<"ROLLBACK;"/utf8>>, Tx),
{event_store_error, gleam@string:inspect(Error@1)}
end
) end
).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 179).
-spec load_events(
sqlite_store(any(), any(), SJP, any()),
sqlight:connection(),
binary(),
integer()
) -> {ok, list(eventsourcing:event_envelop(SJP))} |
{error, eventsourcing:event_sourcing_error(any())}.
load_events(Sqlite_store, Tx, Aggregate_id, Start_from) ->
gleam@result:'try'(
begin
_pipe = sqlight:exec(<<"BEGIN;"/utf8>>, Tx),
gleam@result:map_error(
_pipe,
fun(Error) ->
{event_store_error, gleam@string:inspect(Error)}
end
)
end,
fun(_) ->
Row_decoder = begin
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Aggregate_id@1) ->
gleam@dynamic@decode:field(
2,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Sequence) ->
gleam@dynamic@decode:field(
5,
begin
gleam@dynamic@decode:then(
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Payload_string) ->
_assert_subject = gleam@json:parse(
Payload_string,
erlang:element(
4,
Sqlite_store
)
),
{ok, Payload} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(
#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"load_events"/utf8>>,
line => 197}
)
end,
gleam@dynamic@decode:success(
Payload
)
end
)
end,
fun(Payload@1) ->
gleam@dynamic@decode:field(
6,
metadata_decoder(),
fun(Metadata) ->
gleam@dynamic@decode:field(
4,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Event_type) ->
gleam@dynamic@decode:field(
0,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Event_version) ->
gleam@dynamic@decode:field(
3,
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(
Aggregate_type
) ->
gleam@dynamic@decode:success(
{serialized_event_envelop,
Aggregate_id@1,
Sequence,
Payload@1,
Metadata,
Event_type,
Event_version,
Aggregate_type}
)
end
)
end
)
end
)
end
)
end
)
end
)
end
)
end,
_pipe@1 = sqlight:'query'(
<<"
SELECT aggregate_type, aggregate_id, sequence, event_type, event_version, payload, metadata
FROM event
WHERE aggregate_type = $1
AND aggregate_id = $2
AND sequence > $3
ORDER BY sequence
"/utf8>>,
Tx,
[sqlight:text(erlang:element(7, Sqlite_store)),
sqlight:text(Aggregate_id),
sqlight:int(Start_from)],
Row_decoder
),
gleam@result:map_error(
_pipe@1,
fun(Error@1) ->
{event_store_error, gleam@string:inspect(Error@1)}
end
)
end
).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 140).
-spec create_event_table(sqlite_store(any(), any(), any(), any())) -> {ok, nil} |
{error, eventsourcing:event_sourcing_error(any())}.
create_event_table(Sqlite_store) ->
gleam@result:'try'(
begin
_pipe = sqlight:open(erlang:element(2, Sqlite_store)),
gleam@result:map_error(
_pipe,
fun(Error) ->
{event_store_error,
<<"Failed to open connection: "/utf8,
(gleam@string:inspect(Error))/binary>>}
end
)
end,
fun(Db) ->
_pipe@1 = sqlight:exec(
<<"
CREATE TABLE IF NOT EXISTS event
(
aggregate_type text NOT NULL,
aggregate_id text NOT NULL,
sequence bigint CHECK (sequence >= 0) NOT NULL,
event_type text NOT NULL,
event_version text NOT NULL,
payload text NOT NULL,
metadata text,
PRIMARY KEY (aggregate_type, aggregate_id, sequence)
);
"/utf8>>,
Db
),
gleam@result:map_error(
_pipe@1,
fun(Error@1) ->
{event_store_error,
<<"Failed to create event table: "/utf8,
(gleam@string:inspect(Error@1))/binary>>}
end
)
end
).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 159).
-spec create_snapshot_table(sqlite_store(any(), any(), any(), SJF)) -> {ok, nil} |
{error, eventsourcing:event_sourcing_error(SJF)}.
create_snapshot_table(Sqlite_store) ->
gleam@result:'try'(
begin
_pipe = sqlight:open(erlang:element(2, Sqlite_store)),
gleam@result:map_error(
_pipe,
fun(Error) ->
{event_store_error,
<<"Failed to open connection: "/utf8,
(gleam@string:inspect(Error))/binary>>}
end
)
end,
fun(Db) ->
_pipe@1 = sqlight:exec(
<<"
CREATE TABLE IF NOT EXISTS snapshot
(
aggregate_type text NOT NULL,
aggregate_id text NOT NULL,
sequence bigint CHECK (sequence >= 0) NOT NULL,
entity text NOT NULL,
timestamp int NOT NULL,
PRIMARY KEY (aggregate_type, aggregate_id)
);
"/utf8>>,
Db
),
_pipe@2 = gleam@result:map(_pipe@1, fun(_) -> nil end),
gleam@result:map_error(
_pipe@2,
fun(Error@1) ->
{event_store_error,
<<"Failed to create snapshot table: "/utf8,
(gleam@string:inspect(Error@1))/binary>>}
end
)
end
).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 424).
-spec save_snapshot(
sqlite_store(SLZ, any(), any(), any()),
sqlight:connection(),
eventsourcing:snapshot(SLZ)
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(any())}.
save_snapshot(Sqlite_store, Tx, Snapshot) ->
{snapshot, Aggregate_id, Entity, Sequence, Timestamp} = Snapshot,
_pipe@1 = sqlight:'query'(
<<"
INSERT INTO snapshot (aggregate_type, aggregate_id, sequence, entity, timestamp)
VALUES ($1, $2, $3, $4, $5)
ON CONFLICT (aggregate_type, aggregate_id)
DO UPDATE SET
sequence = EXCLUDED.sequence,
entity = EXCLUDED.entity
"/utf8>>,
Tx,
[sqlight:text(erlang:element(7, Sqlite_store)),
sqlight:text(Aggregate_id),
sqlight:int(Sequence),
sqlight:text(
begin
_pipe = Entity,
(erlang:element(8, Sqlite_store))(_pipe)
end
),
sqlight:int(Timestamp)],
{decoder, fun gleam@dynamic@decode:decode_dynamic/1}
),
_pipe@2 = gleam@result:map_error(
_pipe@1,
fun(Error) -> {event_store_error, gleam@string:inspect(Error)} end
),
gleam@result:map(_pipe@2, fun(_) -> nil end).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 381).
-spec load_snapshot(
sqlite_store(SLP, any(), any(), any()),
sqlight:connection(),
binary()
) -> {ok, gleam@option:option(eventsourcing:snapshot(SLP))} |
{error, eventsourcing:event_sourcing_error(any())}.
load_snapshot(Sqlite_store, Tx, Aggregate_id) ->
Row_decoder = begin
gleam@dynamic@decode:field(
1,
{decoder, fun gleam@dynamic@decode:decode_string/1},
fun(Aggregate_id@1) ->
gleam@dynamic@decode:field(
2,
{decoder, fun gleam@dynamic@decode:decode_int/1},
fun(Sequence) ->
gleam@dynamic@decode:field(
3,
begin
gleam@dynamic@decode:then(
{decoder,
fun gleam@dynamic@decode:decode_string/1},
fun(Entity_string) ->
_assert_subject = gleam@json:parse(
Entity_string,
erlang:element(9, Sqlite_store)
),
{ok, Entity} = case _assert_subject of
{ok, _} -> _assert_subject;
_assert_fail ->
erlang:error(
#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
value => _assert_fail,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"load_snapshot"/utf8>>,
line => 391}
)
end,
gleam@dynamic@decode:success(Entity)
end
)
end,
fun(Entity@1) ->
gleam@dynamic@decode:field(
4,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Timestamp) ->
gleam@dynamic@decode:success(
{snapshot,
Aggregate_id@1,
Entity@1,
Sequence,
Timestamp}
)
end
)
end
)
end
)
end
)
end,
_pipe = sqlight:'query'(
<<"
SELECT aggregate_type, aggregate_id, sequence, entity, timestamp
FROM snapshot
WHERE aggregate_type = $1
AND aggregate_id = $2
"/utf8>>,
Tx,
[sqlight:text(erlang:element(7, Sqlite_store)),
sqlight:text(Aggregate_id)],
Row_decoder
),
_pipe@1 = gleam@result:map(_pipe, fun(Response) -> case Response of
[] ->
none;
[Snapshot | _] ->
{some, Snapshot}
end end),
gleam@result:map_error(
_pipe@1,
fun(Error) -> {event_store_error, gleam@string:inspect(Error)} end
).
-file("/Users/renata-amutio/Projects/renatillas/eventsourcing_sqlite/src/eventsourcing_sqlite.gleam", 92).
-spec new(
binary(),
fun((SID) -> binary()),
gleam@dynamic@decode:decoder(SID),
binary(),
binary(),
binary(),
fun((SIF) -> binary()),
gleam@dynamic@decode:decoder(SIF)
) -> eventsourcing:event_store(sqlite_store(SIF, SIH, SID, SII), SIF, SIH, SID, SII, sqlight:connection()).
new(
Connection_string,
Event_encoder,
Event_decoder,
Event_type,
Event_version,
Aggregate_type,
Entity_encoder,
Entity_decoder
) ->
Eventstore = {sqlite_store,
Connection_string,
Event_encoder,
Event_decoder,
Event_type,
Event_version,
Aggregate_type,
Entity_encoder,
Entity_decoder},
{event_store,
execute_in_transaction(Connection_string),
execute_in_transaction(Connection_string),
execute_in_transaction(Connection_string),
execute_in_transaction(Connection_string),
fun(Tx, Aggregate_id, Events, Metadata) ->
commit_events(Eventstore, Tx, Aggregate_id, Events, Metadata)
end,
fun(Sqlite_store, Tx@1, Aggregate_id@1, Start_from) ->
load_events(Sqlite_store, Tx@1, Aggregate_id@1, Start_from)
end,
fun(Tx@2, Aggregate_id@2) ->
load_snapshot(Eventstore, Tx@2, Aggregate_id@2)
end,
fun(Tx@3, Snapshot) -> save_snapshot(Eventstore, Tx@3, Snapshot) end,
Eventstore}.