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, inline]).
-define(FILEPATH, "src/eventsourcing_sqlite.gleam").
-export([load_events/4, new/8, create_event_table/1, create_snapshot_table/1]).
-export_type([sqlite_store/4]).
-type sqlite_store(IJD, IJE, IJF, IJG) :: {sqlite_store,
binary(),
fun((IJF) -> binary()),
gleam@dynamic@decode:decoder(IJF),
binary(),
binary(),
binary(),
fun((IJD) -> binary()),
gleam@dynamic@decode:decoder(IJD)} |
{gleam_phantom, IJE, IJG}.
-file("src/eventsourcing_sqlite.gleam", 456).
-spec execute_in_transaction(binary()) -> fun((fun((sqlight:connection()) -> INT)) -> INT).
execute_in_transaction(Connection_string) ->
fun(_capture) -> sqlight:with_connection(Connection_string, _capture) end.
-file("src/eventsourcing_sqlite.gleam", 430).
-spec save_snapshot(
sqlite_store(INE, any(), any(), any()),
sqlight:connection(),
eventsourcing:snapshot(INE)
) -> {ok, nil} | {error, eventsourcing:event_sourcing_error(any())}.
save_snapshot(Sqlite_store, Tx, Snapshot) ->
{snapshot, Aggregate_id, Entity, Sequence, Timestamp} = Snapshot,
{Seconds, _} = gleam@time@timestamp:to_unix_seconds_and_nanoseconds(
Timestamp
),
_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(Seconds)],
{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("src/eventsourcing_sqlite.gleam", 387).
-spec load_snapshot(
sqlite_store(IMU, any(), any(), any()),
sqlight:connection(),
binary()
) -> {ok, gleam@option:option(eventsourcing:snapshot(IMU))} |
{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) ->
Entity@1 = case gleam@json:parse(
Entity_string,
erlang:element(9, Sqlite_store)
) of
{ok, Entity} -> Entity;
_assert_fail ->
erlang:error(
#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"load_snapshot"/utf8>>,
line => 397,
value => _assert_fail,
start => 11135,
'end' => 11221,
pattern_start => 11146,
pattern_end => 11156}
)
end,
gleam@dynamic@decode:success(Entity@1)
end
)
end,
fun(Entity@2) ->
gleam@dynamic@decode:field(
4,
{decoder,
fun gleam@dynamic@decode:decode_int/1},
fun(Timestamp) ->
gleam@dynamic@decode:success(
{snapshot,
Aggregate_id@1,
Entity@2,
Sequence,
gleam@time@timestamp:from_unix_seconds(
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("src/eventsourcing_sqlite.gleam", 368).
-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("src/eventsourcing_sqlite.gleam", 300).
-spec persist_events(
sqlite_store(any(), any(), IMJ, any()),
sqlight:connection(),
list(eventsourcing:event_envelop(IMJ))
) -> {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@int:range(
Offset,
Offset + 7,
[],
fun gleam@list:prepend/2
),
gleam@list:map(
_pipe,
fun erlang:integer_to_binary/1
)
end,
<<", $"/utf8>>
))/binary>>/binary,
")"/utf8>>,
Sep = case Placeholders of
<<""/utf8>> ->
<<""/utf8>>;
_ ->
<<", "/utf8>>
end,
{
Aggregate_id@1,
Sequence@1,
Payload@1,
Metadata@1,
Event_type@1,
Event_version@1,
Aggregate_type@1} = case Event of
{serialized_event_envelop,
Aggregate_id,
Sequence,
Payload,
Metadata,
Event_type,
Event_version,
Aggregate_type} -> {
Aggregate_id,
Sequence,
Payload,
Metadata,
Event_type,
Event_version,
Aggregate_type};
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"persist_events"/utf8>>,
line => 325,
value => _assert_fail,
start => 9195,
'end' => 9401,
pattern_start => 9206,
pattern_end => 9393})
end,
New_params = [sqlight:text(Aggregate_type@1),
sqlight:text(Aggregate_id@1),
sqlight:int(Sequence@1),
sqlight:text(Event_type@1),
sqlight:text(Event_version@1),
sqlight:text(
begin
_pipe@1 = Payload@1,
(erlang:element(3, Sqlite_store))(_pipe@1)
end
),
sqlight:text(
begin
_pipe@2 = Metadata@1,
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("src/eventsourcing_sqlite.gleam", 271).
-spec wrap_events(
sqlite_store(any(), any(), ILY, any()),
binary(),
list(ILY),
integer(),
list({binary(), binary()})
) -> list(eventsourcing:event_envelop(ILY)).
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("src/eventsourcing_sqlite.gleam", 230).
-spec commit_events(
sqlite_store(ILC, ILD, ILE, ILF),
sqlight:connection(),
eventsourcing:aggregate(ILC, ILD, ILE, ILF),
list(ILE),
list({binary(), binary()})
) -> {ok, {list(eventsourcing:event_envelop(ILE)), integer()}} |
{error, eventsourcing:event_sourcing_error(ILF)}.
commit_events(Sqlite_store, Tx, Aggregate, Events, Metadata) ->
{aggregate, Aggregate_id, _, Sequence} = Aggregate,
Wrapped_events = wrap_events(
Sqlite_store,
Aggregate_id,
Events,
Sequence,
Metadata
),
Last_event@1 = case gleam@list:last(Wrapped_events) of
{ok, Last_event} -> Last_event;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"commit_events"/utf8>>,
line => 244,
value => _assert_fail,
start => 7021,
'end' => 7074,
pattern_start => 7032,
pattern_end => 7046})
end,
Events@1 = begin
_pipe = persist_events(Sqlite_store, Tx, Wrapped_events),
gleam@result:map(
_pipe,
fun(_) -> {Wrapped_events, erlang:element(3, Last_event@1)} end
)
end,
gleam@result:'try'(
begin
_pipe@1 = sqlight:exec(<<"COMMIT;"/utf8>>, Tx),
gleam@result:map_error(
_pipe@1,
fun(Error) -> {event_store_error, erlang:element(3, Error)} end
)
end,
fun(_) -> _pipe@2 = Events@1,
gleam@result:map_error(
_pipe@2,
fun(Error@1) ->
Result = sqlight:exec(<<"ROLLBACK;"/utf8>>, Tx),
{event_store_error,
<<<<"ROLLING BACK: "/utf8,
(gleam@string:inspect(Error@1))/binary>>/binary,
(case Result of
{ok, _} ->
<<""/utf8>>;
{error, E} ->
<<" - Failed to rollback transaction: "/utf8,
(gleam@string:inspect(E))/binary>>
end)/binary>>}
end
) end
).
-file("src/eventsourcing_sqlite.gleam", 375).
-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) ->
Listmetadata@1 = case gleam@json:parse(
Stringmetadata,
gleam@dynamic@decode:list(
gleam@dynamic@decode:list(
{decoder, fun gleam@dynamic@decode:decode_string/1}
)
)
) of
{ok, Listmetadata} -> Listmetadata;
_assert_fail ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"metadata_decoder"/utf8>>,
line => 377,
value => _assert_fail,
start => 10559,
'end' => 10660,
pattern_start => 10570,
pattern_end => 10586})
end,
_pipe = gleam@list:map(
Listmetadata@1,
fun(Metadata) ->
{Key@1, Val@1} = case Metadata of
[Key, Val] -> {Key, Val};
_assert_fail@1 ->
erlang:error(#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"metadata_decoder"/utf8>>,
line => 381,
value => _assert_fail@1,
start => 10706,
'end' => 10738,
pattern_start => 10717,
pattern_end => 10727})
end,
{Key@1, Val@1}
end
),
gleam@dynamic@decode:success(_pipe)
end
).
-file("src/eventsourcing_sqlite.gleam", 180).
-spec load_events(
sqlite_store(any(), any(), IKU, any()),
sqlight:connection(),
binary(),
integer()
) -> {ok, list(eventsourcing:event_envelop(IKU))} |
{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, erlang:element(3, 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) ->
Payload@1 = case gleam@json:parse(
Payload_string,
erlang:element(
4,
Sqlite_store
)
) of
{ok, Payload} -> Payload;
_assert_fail ->
erlang:error(
#{gleam_error => let_assert,
message => <<"Pattern match failed, no pattern matched the value."/utf8>>,
file => <<?FILEPATH/utf8>>,
module => <<"eventsourcing_sqlite"/utf8>>,
function => <<"load_events"/utf8>>,
line => 198,
value => _assert_fail,
start => 5675,
'end' => 5762,
pattern_start => 5686,
pattern_end => 5697}
)
end,
gleam@dynamic@decode:success(
Payload@1
)
end
)
end,
fun(Payload@2) ->
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@2,
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, erlang:element(3, Error@1)}
end
)
end
).
-file("src/eventsourcing_sqlite.gleam", 93).
-spec new(
binary(),
fun((IJI) -> binary()),
gleam@dynamic@decode:decoder(IJI),
binary(),
binary(),
binary(),
fun((IJK) -> binary()),
gleam@dynamic@decode:decoder(IJK)
) -> eventsourcing:event_store(sqlite_store(IJK, IJM, IJI, IJN), IJK, IJM, IJI, IJN, 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}.
-file("src/eventsourcing_sqlite.gleam", 141).
-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,
(erlang:element(3, 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,
(erlang:element(3, Error@1))/binary>>}
end
)
end
).
-file("src/eventsourcing_sqlite.gleam", 160).
-spec create_snapshot_table(sqlite_store(any(), any(), any(), IKK)) -> {ok, nil} |
{error, eventsourcing:event_sourcing_error(IKK)}.
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,
(erlang:element(3, 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,
(erlang:element(3, Error@1))/binary>>}
end
)
end
).